import os
import re
import time
import json
import psutil
import platform
import subprocess
import numpy as np
from datetime import datetime, timedelta
from collections import defaultdict, deque
import threading
import sys
from typing import Dict, List, Any, Tuple, Union, Optional
import socket
import pickle
from math import sqrt
from functools import partial

# 导入数据库模块
try:
    from mysql_database import KylinDatabase

    MYSQL_AVAILABLE = True
except ImportError:
    #print("警告: mysql_database 模块未找到，将使用本地文件存储")
    MYSQL_AVAILABLE = False


    # 创建一个空的KylinDatabase类作为占位符
    class KylinDatabase:
        def __init__(self, *args, **kwargs):
            pass

        def connect(self):
            return False

        def save_training_data(self, *args, **kwargs):
            return False

        def load_model(self, *args, **kwargs):
            return None

        def save_detection_result(self, *args, **kwargs):
            return False


# 颜色编码（用于评分标记）
class Color:
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    BLUE = '\033[94m'
    MAGENTA = '\033[95m'
    CYAN = '\033[96m'
    RESET = '\033[0m'

    @staticmethod
    def colored(text: str, color: str) -> str:
        """为文本添加颜色编码"""
        return f"{color}{text}{Color.RESET}"


# 纯Python异常检测算法集合
class IsolationForest:
    """孤立森林算法，用于异常检测"""

    def __init__(self, n_estimators: int = 10, max_samples: int = 256, contamination: float = 0.1):
        self.n_estimators = n_estimators  # 树的数量
        self.max_samples = max_samples  # 每个树的样本量
        self.contamination = contamination  # 异常比例
        self.trees: List[Dict] = []  # 树集合
        self.threshold_: Optional[float] = None  # 异常判断阈值

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'IsolationForest':
        """训练模型"""
        X = np.asarray(X)
        n_samples = X.shape[0]
        max_samples = min(self.max_samples, n_samples)

        # 并行构建多棵树（保持原逻辑，使用循环而非多线程避免状态问题）
        self.trees = [self._build_tree(X[np.random.choice(n_samples, max_samples, replace=False)])
                      for _ in range(self.n_estimators)]

        # 计算异常分数并设置阈值
        scores = self.decision_function(X)
        self.threshold_ = np.percentile(scores, 100 * (1 - self.contamination))
        return self

    def _build_tree(self, X: np.ndarray, depth: int = 0, max_depth: Optional[int] = None) -> Dict:
        """递归构建孤立树"""
        if max_depth is None:
            max_depth = int(np.ceil(np.log2(X.shape[0])))

        # 终止条件：达到最大深度或样本量<=1
        if depth >= max_depth or X.shape[0] <= 1:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        # 随机选择特征和分裂值
        n_features = X.shape[1]
        feature_idx = np.random.randint(n_features)
        feature_vals = X[:, feature_idx]
        min_val, max_val = feature_vals.min(), feature_vals.max()

        # 特征值相同无法分裂
        if min_val == max_val:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        split_val = np.random.uniform(min_val, max_val)
        left_mask = feature_vals < split_val

        return {
            'feature': feature_idx,
            'split_val': split_val,
            'left': self._build_tree(X[left_mask], depth + 1, max_depth),
            'right': self._build_tree(X[~left_mask], depth + 1, max_depth),
            'size': X.shape[0]
        }

    def _path_length(self, x: np.ndarray, tree: Dict) -> int:
        """计算样本在树中的路径长度"""
        length = 0
        current = tree
        while current['left'] is not None and current['right'] is not None:
            length += 1
            current = current['left'] if x[current['feature']] < current['split_val'] else current['right']
        return length

    def decision_function(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """计算异常分数"""
        X = np.asarray(X)
        n_samples = X.shape[0]

        # 计算平均路径长度
        avg_lengths = np.mean([[self._path_length(x, tree) for tree in self.trees] for x in X], axis=1)

        # 计算修正系数
        if n_samples <= 1:
            c = 1.0
        else:
            c = 2 * (np.log(n_samples - 1) + 0.5772) - 2 * (n_samples - 1) / n_samples

        # 计算异常分数
        return 2 ** (-avg_lengths / c)

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常（1为异常，0为正常）"""
        if self.threshold_ is None:
            raise ValueError("模型尚未训练，请先调用fit方法")
        scores = self.decision_function(X)
        return np.where(scores > self.threshold_, 1, 0)


class EnhancedIsolationForest:
    """增强版孤立森林算法，支持更多特性"""

    def __init__(self, n_estimators: int = 100, contamination: float = 0.1, max_samples: int = 256):
        self.n_estimators = n_estimators
        self.contamination = contamination
        self.max_samples = max_samples
        self.trees: List[Dict] = []
        self.threshold_: Optional[float] = None
        self.feature_importance_: Optional[np.ndarray] = None

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'EnhancedIsolationForest':
        """训练增强版孤立森林"""
        X = np.asarray(X)
        n_samples = X.shape[0]
        max_samples = min(self.max_samples, n_samples)

        # 构建多棵树
        self.trees = []
        for _ in range(self.n_estimators):
            sample_indices = np.random.choice(n_samples, max_samples, replace=False)
            tree = self._build_tree(X[sample_indices])
            self.trees.append(tree)

        # 计算异常分数并设置阈值
        scores = self.decision_function(X)
        self.threshold_ = np.percentile(scores, 100 * (1 - self.contamination))

        # 计算特征重要性
        self.feature_importance_ = self._calculate_feature_importance(X)

        return self

    def _build_tree(self, X: np.ndarray, depth: int = 0, max_depth: Optional[int] = None) -> Dict:
        """构建孤立树"""
        if max_depth is None:
            max_depth = int(np.ceil(np.log2(X.shape[0])))

        if depth >= max_depth or X.shape[0] <= 1:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        n_features = X.shape[1]
        feature_idx = np.random.randint(n_features)
        feature_vals = X[:, feature_idx]
        min_val, max_val = feature_vals.min(), feature_vals.max()

        if min_val == max_val:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        split_val = np.random.uniform(min_val, max_val)
        left_mask = feature_vals < split_val
        right_mask = ~left_mask

        return {
            'feature_idx': feature_idx,
            'split_val': split_val,
            'left': self._build_tree(X[left_mask], depth + 1, max_depth),
            'right': self._build_tree(X[right_mask], depth + 1, max_depth),
            'size': X.shape[0]
        }

    def _path_length(self, x: np.ndarray, tree: Dict) -> int:
        """计算样本在树中的路径长度"""
        if tree['left'] is None and tree['right'] is None:
            return 0

        if 'feature_idx' not in tree:
            return 0

        feature_val = x[tree['feature_idx']]
        if feature_val < tree['split_val']:
            return 1 + self._path_length(x, tree['left'])
        else:
            return 1 + self._path_length(x, tree['right'])

    def decision_function(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """计算异常分数"""
        X = np.asarray(X)
        scores = np.zeros(len(X))

        for i, x in enumerate(X):
            path_lengths = []
            for tree in self.trees:
                path_length = self._path_length(x, tree)
                path_lengths.append(path_length)

            # 计算平均路径长度
            avg_path_length = np.mean(path_lengths)
            scores[i] = avg_path_length

        return scores

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常"""
        scores = self.decision_function(X)
        return np.where(scores > self.threshold_, 1, 0)

    def _calculate_feature_importance(self, X: np.ndarray) -> np.ndarray:
        """计算特征重要性"""
        n_features = X.shape[1]
        importance = np.zeros(n_features)

        for tree in self.trees:
            if 'feature_idx' in tree:
                feature_idx = tree['feature_idx']
                importance[feature_idx] += 1

        # 归一化
        if importance.sum() > 0:
            importance /= importance.sum()

        return importance

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'IsolationForest':
        """训练模型"""
        X = np.asarray(X)
        n_samples = X.shape[0]
        max_samples = min(self.max_samples, n_samples)

        # 并行构建多棵树（保持原逻辑，使用循环而非多线程避免状态问题）
        self.trees = [self._build_tree(X[np.random.choice(n_samples, max_samples, replace=False)])
                      for _ in range(self.n_estimators)]

        # 计算异常分数并设置阈值
        scores = self.decision_function(X)
        self.threshold_ = np.percentile(scores, 100 * (1 - self.contamination))
        return self

    def _build_tree(self, X: np.ndarray, depth: int = 0, max_depth: Optional[int] = None) -> Dict:
        """递归构建孤立树"""
        if max_depth is None:
            max_depth = int(np.ceil(np.log2(X.shape[0])))

        # 终止条件：达到最大深度或样本量<=1
        if depth >= max_depth or X.shape[0] <= 1:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        # 随机选择特征和分裂值
        n_features = X.shape[1]
        feature_idx = np.random.randint(n_features)
        feature_vals = X[:, feature_idx]
        min_val, max_val = feature_vals.min(), feature_vals.max()

        # 特征值相同无法分裂
        if min_val == max_val:
            return {'left': None, 'right': None, 'size': X.shape[0]}

        split_val = np.random.uniform(min_val, max_val)
        left_mask = feature_vals < split_val

        return {
            'feature': feature_idx,
            'split_val': split_val,
            'left': self._build_tree(X[left_mask], depth + 1, max_depth),
            'right': self._build_tree(X[~left_mask], depth + 1, max_depth),
            'size': X.shape[0]
        }

    def _path_length(self, x: np.ndarray, tree: Dict) -> int:
        """计算样本在树中的路径长度"""
        length = 0
        current = tree
        while current['left'] is not None and current['right'] is not None:
            length += 1
            current = current['left'] if x[current['feature']] < current['split_val'] else current['right']
        return length

    def decision_function(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """计算异常分数"""
        X = np.asarray(X)
        n_samples = X.shape[0]

        # 计算平均路径长度
        avg_lengths = np.mean([[self._path_length(x, tree) for tree in self.trees] for x in X], axis=1)

        # 计算修正系数
        if n_samples <= 1:
            c = 1.0
        else:
            c = 2 * (np.log(n_samples - 1) + 0.5772) - 2 * (n_samples - 1) / n_samples

        # 计算异常分数
        return 2 ** (-avg_lengths / c)

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常（1为异常，0为正常）"""
        if self.threshold_ is None:
            raise ValueError("模型尚未训练，请先调用fit方法")
        scores = self.decision_function(X)
        return np.where(scores > self.threshold_, 1, 0)


class OneClassSVM:
    """单类SVM算法，用于异常检测"""

    def __init__(self, nu: float = 0.5, kernel: str = 'rbf', gamma: Union[str, float] = 'scale'):
        self.nu = nu  # 异常比例参数
        self.kernel = kernel  # 核函数类型（仅支持rbf）
        self.gamma = gamma  # 核系数
        self.alphas: Optional[np.ndarray] = None  # 拉格朗日乘子
        self.support_vectors: Optional[np.ndarray] = None  # 支持向量
        self.b: float = 0.0  # 偏置项

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'OneClassSVM':
        """训练模型"""
        X = np.asarray(X)
        n_samples, n_features = X.shape

        # 仅支持RBF核
        if self.kernel != 'rbf':
            raise NotImplementedError("仅支持rbf核函数")

        # 自动计算gamma值
        if self.gamma == 'scale':
            self.gamma = 1.0 / (n_features * X.var())
        elif not isinstance(self.gamma, float):
            raise ValueError("gamma必须为'scale'或浮点数")

        # 优化核矩阵计算（使用向量运算替代双重循环）
        sq_dists = np.sum(X ** 2, axis=1).reshape(-1, 1) + np.sum(X ** 2, axis=1) - 2 * np.dot(X, X.T)
        K = np.exp(-self.gamma * sq_dists)

        # 简化的训练过程
        self.alphas = np.ones(n_samples) * (1.0 / n_samples)
        self.support_vectors = X
        self.b = -np.percentile(self._decision_function(X), 100 * self.nu)
        return self

    def _rbf_kernel(self, x1: np.ndarray, x2: np.ndarray) -> float:
        """RBF核函数计算"""
        return np.exp(-self.gamma * np.sum((x1 - x2) ** 2))

    def _decision_function(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """决策函数（计算样本到超平面的距离）"""
        X = np.asarray(X)
        n_samples = X.shape[0]
        scores = np.zeros(n_samples)

        # 向量化计算提升效率
        if self.support_vectors is not None and self.alphas is not None:
            for i in range(n_samples):
                scores[i] = np.sum(self.alphas * np.array([self._rbf_kernel(X[i], sv)
                                                           for sv in self.support_vectors])) - self.b
        return scores

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常（1为异常，0为正常）"""
        scores = self._decision_function(X)
        return np.where(scores < 0, 1, 0)


class Autoencoder:
    """自编码器，用于异常检测"""

    def __init__(self, input_dim: int, hidden_dim: int = 8, epochs: int = 10, lr: float = 0.01):
        self.input_dim = input_dim  # 输入维度
        self.hidden_dim = hidden_dim  # 隐藏层维度
        self.epochs = epochs  # 训练轮次
        self.lr = lr  # 学习率
        self.mean: Optional[np.ndarray] = None  # 训练数据均值
        self.std: Optional[np.ndarray] = None  # 训练数据标准差
        self.threshold_: Optional[float] = None  # 异常判断阈值

        # 初始化权重（使用更稳定的初始化方式）
        self.weights1 = np.random.randn(input_dim, hidden_dim) * np.sqrt(2.0 / input_dim)
        self.bias1 = np.zeros(hidden_dim)
        self.weights2 = np.random.randn(hidden_dim, input_dim) * np.sqrt(2.0 / hidden_dim)
        self.bias2 = np.zeros(input_dim)

    def sigmoid(self, x: np.ndarray) -> np.ndarray:
        """sigmoid激活函数 - 防止数值溢出"""
        # 防止数值溢出
        x = np.clip(x, -500, 500)  # 限制输入范围
        return 1.0 / (1.0 + np.exp(-x))

    def sigmoid_deriv(self, x: np.ndarray) -> np.ndarray:
        """sigmoid导数（基于输出值计算）"""
        return x * (1 - x)

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'Autoencoder':
        """训练模型"""
        X = np.asarray(X)
        n_samples = X.shape[0]

        # 数据标准化
        self.mean = np.mean(X, axis=0)
        self.std = np.std(X, axis=0) + 1e-6  # 避免除零
        X_norm = (X - self.mean) / self.std

        # 训练自编码器
        for _ in range(self.epochs):
            # 前向传播
            hidden = self.sigmoid(np.dot(X_norm, self.weights1) + self.bias1)
            output = self.sigmoid(np.dot(hidden, self.weights2) + self.bias2)

            # 计算损失和梯度
            error = X_norm - output
            d_output = error * self.sigmoid_deriv(output)
            d_hidden = np.dot(d_output, self.weights2.T) * self.sigmoid_deriv(hidden)

            # 更新权重（使用均值梯度提升稳定性）
            self.weights2 += self.lr * np.dot(hidden.T, d_output) / n_samples
            self.bias2 += self.lr * np.mean(d_output, axis=0)
            self.weights1 += self.lr * np.dot(X_norm.T, d_hidden) / n_samples
            self.bias1 += self.lr * np.mean(d_hidden, axis=0)

        # 计算重构误差并设置阈值（10%异常率）
        reconstructions = self.reconstruct(X)
        errors = np.mean(np.square(X - reconstructions), axis=1)
        self.threshold_ = np.percentile(errors, 90)
        return self

    def reconstruct(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """重构输入数据"""
        if self.mean is None or self.std is None:
            raise ValueError("模型尚未训练，请先调用fit方法")

        X = np.asarray(X)
        X_norm = (X - self.mean) / self.std
        hidden = self.sigmoid(np.dot(X_norm, self.weights1) + self.bias1)
        output = self.sigmoid(np.dot(hidden, self.weights2) + self.bias2)
        return output * self.std + self.mean

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常（1为异常，0为正常）"""
        if self.threshold_ is None:
            raise ValueError("模型尚未训练，请先调用fit方法")

        reconstructions = self.reconstruct(X)
        errors = np.mean(np.square(X - reconstructions), axis=1)
        return np.where(errors > self.threshold_, 1, 0)


class AdvancedAutoencoder:
    """高级自编码器，支持多层架构"""

    def __init__(self, input_dim: int, hidden_dims: List[int] = None, epochs: int = 30, lr: float = 0.01):
        self.input_dim = input_dim
        self.hidden_dims = hidden_dims or [16, 8, 4]
        self.epochs = epochs
        self.lr = lr

        # 构建编码器和解码器权重
        self.encoder_weights = []
        self.encoder_biases = []
        self.decoder_weights = []
        self.decoder_biases = []

        # 编码器层
        prev_dim = input_dim
        for hidden_dim in self.hidden_dims:
            self.encoder_weights.append(np.random.randn(prev_dim, hidden_dim) * np.sqrt(2.0 / prev_dim))
            self.encoder_biases.append(np.zeros(hidden_dim))
            prev_dim = hidden_dim

        # 解码器层（反向）
        for i in range(len(self.hidden_dims) - 1, -1, -1):
            if i == 0:
                # 最后一层解码到输入维度
                self.decoder_weights.append(
                    np.random.randn(self.hidden_dims[i], input_dim) * np.sqrt(2.0 / self.hidden_dims[i]))
                self.decoder_biases.append(np.zeros(input_dim))
            else:
                # 中间层
                self.decoder_weights.append(
                    np.random.randn(self.hidden_dims[i], self.hidden_dims[i - 1]) * np.sqrt(2.0 / self.hidden_dims[i]))
                self.decoder_biases.append(np.zeros(self.hidden_dims[i - 1]))

        # 数据标准化参数
        self.mean = None
        self.std = None

    def _encode(self, X: np.ndarray, training: bool = False) -> np.ndarray:
        """编码过程"""
        encoded = X
        for i, (weights, bias) in enumerate(zip(self.encoder_weights, self.encoder_biases)):
            encoded = np.dot(encoded, weights) + bias
            encoded = self.sigmoid(encoded)
        return encoded

    def _decode(self, encoded: np.ndarray, training: bool = False) -> np.ndarray:
        """解码过程"""
        decoded = encoded
        for i, (weights, bias) in enumerate(zip(self.decoder_weights, self.decoder_biases)):
            decoded = np.dot(decoded, weights) + bias
            decoded = self.sigmoid(decoded)
        return decoded

    def sigmoid(self, x: np.ndarray) -> np.ndarray:
        """sigmoid激活函数 - 防止数值溢出"""
        x = np.clip(x, -500, 500)  # 限制输入范围
        return 1.0 / (1.0 + np.exp(-x))

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'AdvancedAutoencoder':
        """训练高级自编码器"""
        X = np.asarray(X)
        n_samples = X.shape[0]

        # 数据标准化
        self.mean = np.mean(X, axis=0)
        self.std = np.std(X, axis=0) + 1e-6
        X_norm = (X - self.mean) / self.std

        # 训练循环
        for epoch in range(self.epochs):
            # 前向传播
            encoded = self._encode(X_norm, training=True)
            decoded = self._decode(encoded, training=True)

            # 计算损失
            reconstruction_error = np.mean(np.square(X_norm - decoded))

            # 添加正则化损失
            l2_regularization = 0
            for weights in self.encoder_weights + self.decoder_weights:
                l2_regularization += np.sum(weights ** 2)
            l2_regularization *= 0.0001  # L2正则化系数

            total_loss = reconstruction_error + l2_regularization

            if epoch % 10 == 0:
                print(f"Epoch {epoch}, Loss: {total_loss:.6f}")

        return self

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """预测异常"""
        X = np.asarray(X)
        if self.mean is not None and self.std is not None:
            X_norm = (X - self.mean) / self.std
        else:
            X_norm = X

        encoded = self._encode(X_norm)
        decoded = self._decode(encoded)

        # 计算重构误差作为异常分数
        reconstruction_error = np.mean(np.square(X_norm - decoded), axis=1)

        # 使用阈值判断异常（这里使用简单的统计阈值）
        threshold = np.percentile(reconstruction_error, 95)  # 95%分位数作为阈值
        return np.where(reconstruction_error > threshold, 1, 0)


# 特征工程工具类
class FeatureEngineering:
    """特征工程工具，用于提取和转换多模态特征"""

    @staticmethod
    def sliding_window_features(history: List[float], window_size: int = 10) -> Dict[str, float]:
        """计算滑动窗口内的统计特征"""
        if len(history) < window_size:
            return {}

        window = np.array(history[-window_size:])
        return {
            'window_mean': float(np.mean(window)),
            'window_std': float(np.std(window)),
            'window_max': float(np.max(window)),
            'window_min': float(np.min(window)),
            'window_range': float(np.max(window) - np.min(window)),
            'window_trend': FeatureEngineering._calculate_trend(window)
        }

    @staticmethod
    def _calculate_trend(values: Union[List[float], np.ndarray]) -> float:
        """计算趋势特征（线性回归斜率）"""
        values = np.asarray(values)
        n = len(values)
        if n < 2:
            return 0.0

        x = np.arange(n)
        numerator = n * np.sum(x * values) - np.sum(x) * np.sum(values)
        denominator = n * np.sum(x ** 2) - np.sum(x) ** 2

        return numerator / denominator if denominator != 0 else 0.0

    @staticmethod
    def periodic_feature(current_value: float, historical_values: List[float], period: int = 24) -> float:
        """计算周期性特征（与历史同期的偏差率）"""
        if len(historical_values) < period:
            return 0.0

        historical_value = historical_values[-period]
        return (current_value - historical_value) / (historical_value + 1e-6)  # 避免除零

    @staticmethod
    def tfidf_vectorize(keywords_list: List[List[str]], current_keywords: List[str]) -> Dict[str, float]:
        """对日志关键词进行TF-IDF向量化"""
        # 收集所有关键词
        all_keywords = set(current_keywords)
        for keywords in keywords_list:
            all_keywords.update(keywords)

        keyword_list = list(all_keywords)
        n_keywords = len(keyword_list)
        if n_keywords == 0:
            return {}

        # 计算文档频率（DF）
        df = defaultdict(int)
        total_docs = len(keywords_list) + 1  # 包含当前文档
        for keywords in keywords_list:
            for kw in set(keywords):
                df[kw] += 1

        # 计算当前文档的词频（TF）
        current_tf = defaultdict(int)
        for kw in current_keywords:
            current_tf[kw] += 1
        max_tf = max(current_tf.values()) if current_tf else 1

        # 计算TF-IDF
        return {
            f"tfidf_{kw}": (current_tf.get(kw, 0) / max_tf) * np.log(total_docs / (df.get(kw, 0) + 1))
            for kw in keyword_list
        }

    @staticmethod
    def cross_features(metrics: Dict[str, float]) -> Dict[str, float]:
        """创建特征交叉项"""
        cross = {}
        keys = list(metrics.keys())

        # 基础交叉项（所有特征两两相乘）
        for i in range(len(keys)):
            for j in range(i + 1, len(keys)):
                key1, key2 = keys[i], keys[j]
                cross[f"cross_{key1}_{key2}"] = metrics[key1] * metrics[key2]

        # 业务特定交叉项
        if 'cpu_usage' in metrics and 'mysql_connections' in metrics:
            cross['cpu_mysql_interaction'] = metrics['cpu_usage'] * metrics['mysql_connections']
        if 'memory_usage' in metrics and 'disk_usage' in metrics:
            cross['mem_disk_interaction'] = metrics['memory_usage'] * metrics['disk_usage']
        if 'npu_usage' in metrics and 'cpu_usage' in metrics:
            cross['npu_cpu_interaction'] = metrics['npu_usage'] * metrics['cpu_usage']

        return cross

    @staticmethod
    def pca(X: Union[List[List[float]], np.ndarray], n_components: int = 2) -> np.ndarray:
        """主成分分析降维"""
        X = np.asarray(X)
        X_centered = X - np.mean(X, axis=0)  # 中心化

        # 计算协方差矩阵并分解
        cov_matrix = np.cov(X_centered, rowvar=False)
        eigenvalues, eigenvectors = np.linalg.eigh(cov_matrix)

        # 按特征值排序并选择前n个主成分
        sorted_indices = np.argsort(eigenvalues)[::-1]
        top_eigenvectors = eigenvectors[:, sorted_indices[:n_components]]

        return np.dot(X_centered, top_eigenvectors)


# 纯Python LSTM模型实现（用于时序预测）
class SimpleLSTM:
    """轻量级LSTM模型，用于系统指标趋势预测"""

    def __init__(self, input_size: int = 1, hidden_size: int = 8, output_size: int = 1):
        self.input_size = input_size  # 输入维度
        self.hidden_size = hidden_size  # 隐藏层维度
        self.output_size = output_size  # 输出维度

        # 初始化权重（使用Xavier初始化提升稳定性）
        scale = np.sqrt(1.0 / hidden_size)
        self.Wf = np.random.randn(hidden_size, input_size + hidden_size) * scale
        self.bf = np.zeros((hidden_size, 1))
        self.Wi = np.random.randn(hidden_size, input_size + hidden_size) * scale
        self.bi = np.zeros((hidden_size, 1))
        self.Wc = np.random.randn(hidden_size, input_size + hidden_size) * scale
        self.bc = np.zeros((hidden_size, 1))
        self.Wo = np.random.randn(hidden_size, input_size + hidden_size) * scale
        self.bo = np.zeros((hidden_size, 1))
        self.Wy = np.random.randn(output_size, hidden_size) * scale
        self.by = np.zeros((output_size, 1))

        # 动量优化参数
        self.learning_rate = 0.001
        self.momentum = 0.9
        self.velocities = {
            'Wf': np.zeros_like(self.Wf),
            'bf': np.zeros_like(self.bf),
            'Wi': np.zeros_like(self.Wi),
            'bi': np.zeros_like(self.bi),
            'Wc': np.zeros_like(self.Wc),
            'bc': np.zeros_like(self.bc),
            'Wo': np.zeros_like(self.Wo),
            'bo': np.zeros_like(self.bo),
            'Wy': np.zeros_like(self.Wy),
            'by': np.zeros_like(self.by)
        }

    def sigmoid(self, x: np.ndarray) -> np.ndarray:
        """sigmoid激活函数 - 防止数值溢出"""
        # 防止数值溢出
        x = np.clip(x, -500, 500)  # 限制输入范围
        return 1.0 / (1.0 + np.exp(-x))

    def tanh(self, x: np.ndarray) -> np.ndarray:
        """tanh激活函数"""
        return np.tanh(x)

    def forward(self, x: Union[float, List[float], np.ndarray],
                h_prev: np.ndarray, c_prev: np.ndarray) -> Tuple[np.ndarray, ...]:
        """前向传播计算"""
        x = np.asarray(x).reshape(-1, 1)  # 确保输入形状正确
        combined = np.concatenate((h_prev, x), axis=0)  # 拼接隐藏状态和输入

        # LSTM门计算
        ft = self.sigmoid(np.dot(self.Wf, combined) + self.bf)  # 遗忘门
        it = self.sigmoid(np.dot(self.Wi, combined) + self.bi)  # 输入门
        cct = self.tanh(np.dot(self.Wc, combined) + self.bc)  # 细胞候选值
        c_next = ft * c_prev + it * cct  # 细胞状态
        ot = self.sigmoid(np.dot(self.Wo, combined) + self.bo)  # 输出门
        h_next = ot * self.tanh(c_next)  # 隐藏状态
        y = np.dot(self.Wy, h_next) + self.by  # 输出

        return y, h_next, c_next, ft, it, cct, ot

    def train_step(self, sequence: List[Union[float, List[float]]], target: float) -> float:
        """单步训练，用于增量学习"""
        h = np.zeros((self.hidden_size, 1))
        c = np.zeros((self.hidden_size, 1))
        forward_cache = []

        # 前向传播
        for x in sequence:
            y, h, c, ft, it, cct, ot = self.forward(x, h, c)
            forward_cache.append((x, h, c, ft, it, cct, ot))

        # 计算输出层误差
        dy = y - target
        dWy = np.dot(dy, forward_cache[-1][1].T)
        dby = dy

        # 反向传播初始化
        dh_next = np.dot(self.Wy.T, dy)
        dc_next = np.zeros_like(forward_cache[-1][2])
        gradients = {
            'Wf': np.zeros_like(self.Wf),
            'bf': np.zeros_like(self.bf),
            'Wi': np.zeros_like(self.Wi),
            'bi': np.zeros_like(self.bi),
            'Wc': np.zeros_like(self.Wc),
            'bc': np.zeros_like(self.bc),
            'Wo': np.zeros_like(self.Wo),
            'bo': np.zeros_like(self.bo),
            'Wy': dWy,
            'by': dby
        }

        # 反向遍历序列
        c_prev = np.zeros((self.hidden_size, 1))
        for t in reversed(range(len(forward_cache))):
            x, h, c, ft, it, cct, ot = forward_cache[t]
            h_prev = forward_cache[t - 1][1] if t > 0 else np.zeros((self.hidden_size, 1))

            # 计算梯度
            dh = dh_next + dc_next * ot * (1 - self.tanh(c) ** 2)
            dot = dh * self.tanh(c) * ot * (1 - ot)
            dcct = dc_next * it * (1 - cct ** 2) + it * (1 - it) * cct * dh_next
            dft = dc_next * c_prev * ft * (1 - ft)
            dit = dcct * cct * it * (1 - it)

            combined = np.concatenate((h_prev, np.asarray(x).reshape(-1, 1)), axis=0)
            gradients['Wf'] += np.dot(dft, combined.T)
            gradients['bf'] += dft
            gradients['Wi'] += np.dot(dit, combined.T)
            gradients['bi'] += dit
            gradients['Wc'] += np.dot(dcct, combined.T)
            gradients['bc'] += dcct
            gradients['Wo'] += np.dot(dot, combined.T)
            gradients['bo'] += dot

            # 更新反向传播状态
            dh_prev = (np.dot(self.Wf[:, :self.hidden_size].T, dft) +
                       np.dot(self.Wi[:, :self.hidden_size].T, dit) +
                       np.dot(self.Wc[:, :self.hidden_size].T, dcct) +
                       np.dot(self.Wo[:, :self.hidden_size].T, dot))
            dc_prev = dc_next * ft + ot * (1 - self.tanh(c) ** 2) * dc_next * ft

            dh_next, dc_next, c_prev = dh_prev, dc_prev, c

        # 带动量的权重更新
        for key in gradients:
            self.velocities[key] = self.momentum * self.velocities[key] - self.learning_rate * gradients[key]
            setattr(self, key, getattr(self, key) + self.velocities[key])

        return float(np.mean(dy ** 2))

    def predict(self, sequence: List[Union[float, List[float]]], steps: int = 5) -> List[float]:
        """预测未来steps步的值"""
        if not sequence:
            return []

        # 初始化隐藏状态
        h = np.zeros((self.hidden_size, 1))
        c = np.zeros((self.hidden_size, 1))

        # 前向传播获取最终状态
        for x in sequence:
            y, h, c, _, _, _, _ = self.forward(x, h, c)

        # 预测未来步骤
        predictions = []
        current_input = y
        for _ in range(steps):
            current_input = np.asarray(current_input).reshape(-1, 1)
            y, h, c, _, _, _, _ = self.forward(current_input, h, c)
            predictions.append(float(y[0, 0]))
            current_input = y

        return predictions


# 无监督异常检测模块（基于密度聚类）
class SimpleDBSCAN:
    """简化版DBSCAN算法，用于自动发现异常模式"""

    def __init__(self, eps: float = 0.5, min_samples: int = 5):
        self.eps = eps  # 邻域半径
        self.min_samples = min_samples  # 最小样本数
        self.labels: Optional[np.ndarray] = None  # 聚类标签（-1为异常）

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'SimpleDBSCAN':
        """训练模型（聚类）"""
        X = np.asarray(X)
        self.labels = np.full(len(X), -1, dtype=int)  # 初始化标签为-1（未分类）
        cluster_id = 0

        for i in range(len(X)):
            if self.labels[i] != -1:  # 已分类样本跳过
                continue

            neighbors = self._find_neighbors(X, i)
            if len(neighbors) < self.min_samples:  # 核心点判断
                continue

            # 扩展聚类
            self._expand_cluster(X, i, neighbors, cluster_id)
            cluster_id += 1

        return self

    def _find_neighbors(self, X: np.ndarray, idx: int) -> List[int]:
        """查找指定样本的邻域样本"""
        return [i for i in range(len(X))
                if np.linalg.norm(X[idx] - X[i]) < self.eps]

    def _expand_cluster(self, X: np.ndarray, idx: int, neighbors: List[int], cluster_id: int) -> None:
        """扩展聚类簇"""
        self.labels[idx] = cluster_id
        i = 0
        while i < len(neighbors):
            point_idx = neighbors[i]
            if self.labels[point_idx] == -1:  # 未分类样本
                self.labels[point_idx] = cluster_id
                new_neighbors = self._find_neighbors(X, point_idx)
                if len(new_neighbors) >= self.min_samples:  # 核心点的邻域合并
                    neighbors.extend(new_neighbors)
            i += 1


class EnsembleAnomalyDetector:
    """集成异常检测器 - 组合多个模型"""

    def __init__(self, models: Dict[str, Any] = None):
        self.models = models or {}
        self.model_weights = {}
        self.ensemble_threshold = 0.5

    def add_model(self, name: str, model: Any, weight: float = 1.0):
        """添加模型到集成"""
        self.models[name] = model
        self.model_weights[name] = weight

    def fit(self, X: Union[List[List[float]], np.ndarray]) -> 'EnsembleAnomalyDetector':
        """训练所有模型"""
        X = np.asarray(X)

        for name, model in self.models.items():
            print(f"训练模型: {name}")
            if hasattr(model, 'fit'):
                model.fit(X)

        return self

    def predict(self, X: Union[List[List[float]], np.ndarray]) -> np.ndarray:
        """集成预测"""
        X = np.asarray(X)
        predictions = {}
        scores = {}

        # 获取每个模型的预测
        for name, model in self.models.items():
            if hasattr(model, 'predict'):
                predictions[name] = model.predict(X)
            if hasattr(model, 'decision_function'):
                scores[name] = model.decision_function(X)

        # 加权投票
        ensemble_scores = np.zeros(X.shape[0])
        total_weight = 0

        for name, score in scores.items():
            weight = self.model_weights.get(name, 1.0)
            ensemble_scores += score * weight
            total_weight += weight

        if total_weight > 0:
            ensemble_scores /= total_weight

        # 基于阈值判断异常
        return np.where(ensemble_scores > self.ensemble_threshold, 1, 0)


# 分布式调用链追踪器
class DistributedCallTracer:
    """分布式调用链追踪器，支持跨节点调用链记录与分析"""

    def __init__(self):
        self.call_chains = deque(maxlen=1000)  # 调用链历史（限制最大长度）
        self.current_chain: Optional[Dict] = None  # 当前活跃调用链
        self.node_id = socket.gethostname()  # 当前节点标识

    def start_chain(self, service: str, operation: str) -> str:
        """开始新的调用链"""
        trace_id = f"{self.node_id}-{int(time.time() * 1000)}-{np.random.randint(1000)}"
        self.current_chain = {
            'trace_id': trace_id,
            'start_time': time.time(),
            'nodes': [self.node_id],
            'calls': [{
                'service': service,
                'operation': operation,
                'node': self.node_id,
                'start_time': time.time(),
                'end_time': None,
                'status': 'running'
            }]
        }
        return trace_id

    def add_call(self, trace_id: str, service: str, operation: str, target_node: str) -> None:
        """添加跨节点调用记录"""
        # 查找指定调用链
        if not self.current_chain or self.current_chain['trace_id'] != trace_id:
            for chain in self.call_chains:
                if chain['trace_id'] == trace_id:
                    self.current_chain = chain
                    break
            if not self.current_chain:
                return

        # 添加新调用记录
        self.current_chain['calls'].append({
            'service': service,
            'operation': operation,
            'node': self.node_id,
            'target_node': target_node,
            'start_time': time.time(),
            'end_time': None,
            'status': 'running'
        })

        # 记录参与节点
        if target_node not in self.current_chain['nodes']:
            self.current_chain['nodes'].append(target_node)

    def end_call(self, trace_id: str, status: str = 'success') -> None:
        """结束当前调用"""
        if not self.current_chain or self.current_chain['trace_id'] != trace_id:
            return

        # 更新最后一个调用状态
        if self.current_chain['calls']:
            self.current_chain['calls'][-1]['end_time'] = time.time()
            self.current_chain['calls'][-1]['status'] = status

        # 所有调用结束则完成整个调用链
        if all(c['end_time'] is not None for c in self.current_chain['calls']):
            self.current_chain['end_time'] = time.time()
            self.current_chain['duration'] = self.current_chain['end_time'] - self.current_chain['start_time']
            self.call_chains.append(self.current_chain)
            self.current_chain = None

    def get_slow_calls(self, threshold: float = 1.0) -> List[Dict]:
        """获取慢调用（超过阈值的调用）"""
        slow_calls = []
        for chain in self.call_chains:
            for call in chain['calls']:
                if call['end_time'] and (call['end_time'] - call['start_time'] > threshold):
                    slow_calls.append({
                        'trace_id': chain['trace_id'],
                        'service': call['service'],
                        'operation': call['operation'],
                        'nodes': f"{call['node']}->{call.get('target_node', 'local')}",
                        'duration': call['end_time'] - call['start_time'],
                        'start_time': call['start_time']
                    })
        return sorted(slow_calls, key=lambda x: x['duration'], reverse=True)

    def get_failed_calls(self) -> List[Dict]:
        """获取失败的调用"""
        failed_calls = []
        for chain in self.call_chains:
            for call in chain['calls']:
                if call['status'] != 'success':
                    failed_calls.append({
                        'trace_id': chain['trace_id'],
                        'service': call['service'],
                        'operation': call['operation'],
                        'nodes': f"{call['node']}->{call.get('target_node', 'local')}",
                        'status': call['status'],
                        'start_time': call['start_time']
                    })
        return failed_calls

    def serialize_chain(self, trace_id: str) -> bytes:
        """序列化调用链用于网络传输"""
        try:
            # 查找调用链
            for chain in self.call_chains:
                if chain['trace_id'] == trace_id:
                    return pickle.dumps(chain)
            if self.current_chain and self.current_chain['trace_id'] == trace_id:
                return pickle.dumps(self.current_chain)
        except Exception as e:
            print(Color.colored(f"序列化调用链失败: {e}", Color.RED))
        return b''

    def deserialize_chain(self, data: bytes) -> None:
        """反序列化并合并远程调用链"""
        try:
            chain = pickle.loads(data)
            # 合并已有调用链
            for i, existing in enumerate(self.call_chains):
                if existing['trace_id'] == chain['trace_id']:
                    self.call_chains[i] = self._merge_chains(existing, chain)
                    return
            self.call_chains.append(chain)
        except Exception as e:
            print(Color.colored(f"反序列化调用链失败: {e}", Color.RED))

    def _merge_chains(self, chain1: Dict, chain2: Dict) -> Dict:
        """合并两个相同trace_id的调用链"""
        merged = chain1.copy()

        # 合并调用记录（去重）
        call_key = lambda c: f"{c['service']}-{c['operation']}-{c['start_time']}"
        calls = {call_key(c): c for c in chain1['calls']}
        calls.update({call_key(c): c for c in chain2['calls']})
        merged['calls'] = list(calls.values())

        # 合并节点
        merged['nodes'] = list(set(chain1['nodes'] + chain2['nodes']))

        # 更新时间信息
        merged['start_time'] = min(chain1['start_time'], chain2['start_time'])
        if 'end_time' in chain1 and 'end_time' in chain2:
            merged['end_time'] = max(chain1['end_time'], chain2['end_time'])
            merged['duration'] = merged['end_time'] - merged['start_time']

        return merged


# 模型训练器 - 独立训练和保存异常检测模型
class ModelTrainer:
    """模型训练器 - 独立训练和保存异常检测模型"""

    def __init__(self, model_dir: str = "trained_models"):
        self.model_dir = model_dir
        self.ensure_model_directory()
        self.training_data = []
        self.model_versions = {}

    def ensure_model_directory(self):
        """确保模型目录存在"""
        if not os.path.exists(self.model_dir):
            os.makedirs(self.model_dir)
            print(f"{Color.CYAN}创建模型目录: {self.model_dir}{Color.RESET}")

    def collect_training_data(self, metrics: Dict, duration_hours: int = 1):
        """收集训练数据"""
        features = self._extract_features_for_training(metrics)
        features['timestamp'] = time.time()
        self.training_data.append(features)

        # 保持指定时间范围的数据
        cutoff_time = time.time() - (duration_hours * 3600)
        self.training_data = [data for data in self.training_data if data['timestamp'] >= cutoff_time]

        print(f"{Color.BLUE}收集训练数据: {len(self.training_data)} 条记录{Color.RESET}")

    def _extract_features_for_training(self, metrics: Dict) -> Dict:
        """为训练提取特征（与model_trainer_optimized.py保持一致）"""
        features = {}

        # 基础指标特征（固定特征集，确保维度一致）
        features['cpu_usage'] = metrics['cpu']['usage']
        features['cpu_user'] = metrics['cpu'].get('user', 0)
        features['cpu_system'] = metrics['cpu'].get('system', 0)
        features['cpu_idle'] = metrics['cpu'].get('idle', 0)
        features['cpu_iowait'] = metrics['cpu'].get('iowait', 0)
        features['cpu_irq'] = metrics['cpu'].get('irq', 0)
        features['cpu_softirq'] = metrics['cpu'].get('softirq', 0)
        features['cpu_steal'] = metrics['cpu'].get('steal', 0)
        features['cpu_guest'] = metrics['cpu'].get('guest', 0)
        features['memory_usage'] = metrics['memory']['usage']
        features['memory_available'] = metrics['memory'].get('available', 0)
        features['memory_used'] = metrics['memory'].get('used', 0)
        features['memory_free'] = metrics['memory'].get('free', 0)
        features['memory_cached'] = metrics['memory'].get('cached', 0)
        features['memory_buffers'] = metrics['memory'].get('buffers', 0)
        features['memory_swap_used'] = metrics['memory'].get('swap_used', 0)
        features['disk_usage'] = metrics['disk']['usage']
        features['disk_read_bytes'] = metrics['disk'].get('read_bytes', 0)
        features['disk_write_bytes'] = metrics['disk'].get('write_bytes', 0)
        features['disk_read_count'] = metrics['disk'].get('read_count', 0)
        features['disk_write_count'] = metrics['disk'].get('write_count', 0)
        features['disk_read_time'] = metrics['disk'].get('read_time', 0)
        features['disk_write_time'] = metrics['disk'].get('write_time', 0)
        features['network_in'] = metrics['network']['recv_rate']
        features['network_out'] = metrics['network']['sent_rate']
        features['network_packets_in'] = metrics['network'].get('packets_recv', 0)
        features['network_packets_out'] = metrics['network'].get('packets_sent', 0)
        features['network_err_in'] = metrics['network'].get('errin', 0)
        features['network_err_out'] = metrics['network'].get('errout', 0)
        features['network_drop_in'] = metrics['network'].get('dropin', 0)
        features['network_drop_out'] = metrics['network'].get('dropout', 0)

        # 数据库详细指标特征
        features['mysql_connections'] = metrics['services']['mysql']['connections']
        features['mysql_slow_queries'] = metrics['services']['mysql']['slow_queries']
        features['mysql_threads_connected'] = metrics['services']['mysql'].get('threads_connected', 0)
        features['mysql_threads_running'] = metrics['services']['mysql'].get('threads_running', 0)
        features['mysql_uptime'] = metrics['services']['mysql'].get('uptime', 0)
        features['mysql_questions'] = metrics['services']['mysql'].get('questions', 0)
        features['mysql_opens'] = metrics['services']['mysql'].get('opens', 0)
        features['mysql_flush_commands'] = metrics['services']['mysql'].get('flush_commands', 0)

        # Redis详细指标特征
        features['redis_connected_clients'] = metrics['services']['redis'].get('connected_clients', 0)
        features['redis_used_memory'] = metrics['services']['redis'].get('used_memory', 0)
        features['redis_total_commands'] = metrics['services']['redis'].get('total_commands_processed', 0)
        features['redis_keyspace_hits'] = metrics['services']['redis'].get('keyspace_hits', 0)
        features['redis_keyspace_misses'] = metrics['services']['redis'].get('keyspace_misses', 0)
        features['redis_evicted_keys'] = metrics['services']['redis'].get('evicted_keys', 0)
        features['redis_expired_keys'] = metrics['services']['redis'].get('expired_keys', 0)

        # Web服务详细指标特征
        features['nginx_active_connections'] = metrics['services']['nginx'].get('active_connections', 0)
        features['nginx_5xx_errors'] = metrics['services']['nginx'].get('5xx_errors', 0)
        features['nginx_requests_per_second'] = metrics['services']['nginx'].get('requests_per_second', 0)
        features['nginx_4xx_errors'] = metrics['services']['nginx'].get('4xx_errors', 0)
        features['nginx_total_requests'] = metrics['services']['nginx'].get('total_requests', 0)

        features['apache_5xx_errors'] = metrics['services']['apache'].get('5xx_errors', 0)
        features['apache_requests_per_second'] = metrics['services']['apache'].get('requests_per_second', 0)

        # 系统日志相关特征
        features['error_log_count'] = metrics.get('error_log_count', 0)
        features['warning_log_count'] = metrics.get('warning_log_count', 0)
        features['critical_log_count'] = metrics.get('critical_log_count', 0)

        # 服务状态特征（转换为数值）
        features['mysql_running'] = 1 if metrics['services']['mysql']['running'] else 0
        features['redis_running'] = 1 if metrics['services']['redis']['running'] else 0
        features['nginx_running'] = 1 if metrics['services']['nginx']['running'] else 0
        features['apache_running'] = 1 if metrics['services']['apache']['running'] else 0

        # 添加NPU特征（如果可用）- 限制为最多2个NPU设备
        if metrics['npu']['available']:
            for i, device in enumerate(metrics['npu']['devices'][:2]):  # 最多2个NPU设备
                features[f'npu_{i}_usage'] = device['usage']
                features[f'npu_{i}_temp'] = device['temperature']
                features[f'npu_{i}_mem_usage'] = device['memory_usage']
                features[f'npu_{i}_power'] = device.get('power_usage', 0.0)
                features[f'npu_{i}_frequency'] = device.get('frequency', 0.0)
        else:
            # 如果没有NPU，添加默认值
            for i in range(2):
                features[f'npu_{i}_usage'] = 0.0
                features[f'npu_{i}_temp'] = 0.0
                features[f'npu_{i}_mem_usage'] = 0.0
                features[f'npu_{i}_power'] = 0.0
                features[f'npu_{i}_frequency'] = 0.0

        # 滑动窗口特征（限制数量）
        for key in ['cpu_usage', 'memory_usage', 'disk_usage']:
            # 使用当前值作为默认值（因为没有历史数据）
            features[f"{key}_window_mean"] = features[key]
            features[f"{key}_window_std"] = 0.0
            features[f"{key}_window_max"] = features[key]
            features[f"{key}_window_min"] = features[key]
            features[f"{key}_window_range"] = 0.0
            features[f"{key}_window_trend"] = 0.0

        # 周期性特征（限制数量）
        for key in ['cpu_usage', 'memory_usage']:
            features[f"{key}_periodic_deviation"] = 0.0

        # 交叉特征（限制数量，避免特征爆炸）
        # 只保留最重要的交叉特征
        features['cross_cpu_usage_memory_usage'] = features['cpu_usage'] * features['memory_usage']
        features['cross_cpu_usage_disk_usage'] = features['cpu_usage'] * features['disk_usage']
        features['cross_memory_usage_disk_usage'] = features['memory_usage'] * features['disk_usage']
        features['cross_mysql_connections_redis_connected_clients'] = features['mysql_connections'] * features[
            'redis_connected_clients']
        features['cross_nginx_active_connections_apache_requests_per_second'] = features['nginx_active_connections'] * \
                                                                                features['apache_requests_per_second']

        return features

    def has_sufficient_data(self, min_samples: int = 720) -> bool:  # 1小时数据
        """检查是否有足够的训练数据"""
        return len(self.training_data) >= min_samples

    def train_models(self, model_version: str = None) -> Dict:
        """训练所有异常检测模型"""
        if not self.has_sufficient_data():
            raise ValueError(f"训练数据不足，需要至少720条记录（1小时），当前只有{len(self.training_data)}条")

        if model_version is None:
            model_version = datetime.now().strftime("%Y%m%d_%H%M%S")

        print(f"{Color.CYAN}开始训练模型版本: {model_version}{Color.RESET}")

        # 准备训练数据
        X = []
        for data in self.training_data:
            features = [v for k, v in data.items() if k != 'timestamp' and isinstance(v, (int, float))]
            X.append(features)

        X = np.array(X)

        # 训练模型
        models = {}

        # 1. 训练孤立森林
        print("训练孤立森林模型...")
        isolation_forest = IsolationForest(n_estimators=10, contamination=0.1)
        isolation_forest.fit(X)
        models['isolation_forest'] = isolation_forest

        # 2. 训练单类SVM
        print("训练单类SVM模型...")
        one_class_svm = OneClassSVM(nu=0.1)
        one_class_svm.fit(X)
        models['one_class_svm'] = one_class_svm

        # 3. 训练自编码器
        print("训练自编码器模型...")
        autoencoder = Autoencoder(input_dim=X.shape[1], hidden_dim=8, epochs=20)
        autoencoder.fit(X)
        models['autoencoder'] = autoencoder

        # 4. 训练LSTM预测模型
        print("训练LSTM预测模型...")
        lstm_model = SimpleLSTM(input_size=X.shape[1], hidden_size=16)
        # 使用时间序列数据训练LSTM
        for i in range(len(X) - 10):
            sequence = X[i:i + 10].tolist()
            target = X[i + 10][0] if i + 10 < len(X) else X[-1][0]  # 预测CPU使用率
            lstm_model.train_step(sequence, target)
        models['lstm'] = lstm_model

        # 保存模型
        self._save_models(models, model_version)

        # 记录模型版本
        self.model_versions[model_version] = {
            'timestamp': time.time(),
            'data_count': len(self.training_data),
            'features_count': X.shape[1]
        }

        print(f"{Color.GREEN}模型训练完成！版本: {model_version}{Color.RESET}")
        return models

    def _save_models(self, models: Dict, version: str):
        """保存训练好的模型"""
        for model_name, model in models.items():
            model_path = os.path.join(self.model_dir, f"{model_name}_{version}.pkl")
            try:
                with open(model_path, 'wb') as f:
                    pickle.dump(model, f)
                print(f"保存模型: {model_path}")
            except Exception as e:
                print(f"{Color.RED}保存模型 {model_name} 失败: {e}{Color.RESET}")

    def load_models(self, version: str = None) -> Dict:
        """加载训练好的模型"""
        if version is None:
            # 加载最新版本
            model_files = [f for f in os.listdir(self.model_dir) if f.endswith('.pkl')]
            if not model_files:
                raise FileNotFoundError("没有找到训练好的模型")

            # 按时间戳排序，取最新版本
            versions = set()
            for file in model_files:
                parts = file.split('_')
                if len(parts) >= 2:
                    versions.add('_'.join(parts[1:]).replace('.pkl', ''))

            if not versions:
                raise FileNotFoundError("无法确定模型版本")

            version = sorted(versions)[-1]

        models = {}
        model_types = ['isolation_forest', 'one_class_svm', 'autoencoder', 'lstm']

        for model_type in model_types:
            model_path = os.path.join(self.model_dir, f"{model_type}_{version}.pkl")
            if os.path.exists(model_path):
                try:
                    with open(model_path, 'rb') as f:
                        models[model_type] = pickle.load(f)
                    print(f"加载模型: {model_path}")
                except Exception as e:
                    print(f"{Color.RED}加载模型 {model_type} 失败: {e}{Color.RESET}")

        return models

    def get_model_info(self) -> Dict:
        """获取模型信息"""
        model_files = [f for f in os.listdir(self.model_dir) if f.endswith('.pkl')]
        versions = {}

        for file in model_files:
            parts = file.split('_')
            if len(parts) >= 2:
                model_type = parts[0]
                version = '_'.join(parts[1:]).replace('.pkl', '')

                if version not in versions:
                    versions[version] = {}

                file_path = os.path.join(self.model_dir, file)
                file_stat = os.stat(file_path)
                versions[version][model_type] = {
                    'file_size': file_stat.st_size,
                    'modified_time': file_stat.st_mtime
                }

        return versions

    def cleanup_old_models(self, keep_versions: int = 3):
        """清理旧模型，保留最新的几个版本"""
        versions = list(self.get_model_info().keys())
        if len(versions) <= keep_versions:
            return

        # 按时间排序，删除旧版本
        versions.sort()
        old_versions = versions[:-keep_versions]

        for version in old_versions:
            for model_type in ['isolation_forest', 'one_class_svm', 'autoencoder', 'lstm']:
                model_path = os.path.join(self.model_dir, f"{model_type}_{version}.pkl")
                if os.path.exists(model_path):
                    os.remove(model_path)
                    print(f"删除旧模型: {model_path}")


# 分布式任务调度器（主从模式）
class DistributedScheduler:
    """分布式任务调度器，支持跨机器检测与风险联动，集成调用链追踪"""

    def __init__(self, master_host: str = '0.0.0.0', master_port: int = 9999):
        self.master_host = master_host
        self.master_port = master_port
        self.slaves: List[Tuple[str, int]] = []  # 从节点列表
        self.results = defaultdict(list)  # 任务结果：{节点IP: 结果列表}
        self.server_thread: Optional[threading.Thread] = None
        self.running = False
        self.dependency_graph: Dict[str, List[str]] = {}  # 机器依赖关系图
        self.task_queue = deque()  # 任务队列
        self.lock = threading.Lock()  # 线程安全锁
        self.call_tracer = DistributedCallTracer()  # 调用链追踪器

    def set_dependency_graph(self, graph: Dict[str, List[str]]) -> None:
        """设置机器依赖关系图，用于风险传播链分析"""
        self.dependency_graph = graph

    def start_master(self) -> None:
        """启动主节点服务"""
        self.running = True
        self.server_thread = threading.Thread(target=self._master_server, daemon=True)
        self.server_thread.start()
        print(Color.colored(f"分布式主节点启动: {self.master_host}:{self.master_port}", Color.BLUE))
        # 启动任务处理线程
        threading.Thread(target=self._process_tasks, daemon=True).start()

    def _master_server(self) -> None:
        """主节点服务端，接收从节点结果"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.bind((self.master_host, self.master_port))
                s.listen(5)
                s.settimeout(1.0)  # 超时控制，便于退出

                while self.running:
                    try:
                        conn, addr = s.accept()
                        with conn:
                            data = conn.recv(4096)
                            if data:
                                try:
                                    task_result = pickle.loads(data)
                                    # 处理调用链数据
                                    if 'call_chain' in task_result:
                                        self.call_tracer.deserialize_chain(task_result['call_chain'])
                                        del task_result['call_chain']

                                    # 保存结果
                                    with self.lock:
                                        self.results[addr[0]].append(task_result)

                                    # 风险联动：高风险时通知依赖节点
                                    if 'risk_level' in task_result and task_result['risk_level'] >= 7:
                                        self._propagate_risk(addr[0], task_result)

                                    conn.sendall(b"ACK")
                                except Exception as e:
                                    conn.sendall(f"ERROR: {str(e)}".encode())
                    except socket.timeout:
                        continue  # 超时继续循环
                    except Exception as e:
                        if self.running:
                            print(Color.colored(f"主节点错误: {e}", Color.RED))
        except Exception as e:
            if self.running:
                print(Color.colored(f"主节点服务启动失败: {e}", Color.RED))

    def _process_tasks(self) -> None:
        """处理任务队列，实现动态负载均衡"""
        while self.running:
            if self.task_queue and self.slaves:
                task = self.task_queue.popleft()

                # 负载均衡：选择任务最少的从节点
                with self.lock:
                    slave_load = {slave: len(self.results.get(slave[0], [])) for slave in self.slaves}
                least_loaded = min(slave_load, key=slave_load.get)

                # 记录调用链
                trace_id = self.call_tracer.start_chain(
                    service="task_scheduler",
                    operation="distributed_task"
                )
                self.call_tracer.add_call(
                    trace_id=trace_id,
                    service="task_scheduler",
                    operation="execute_task",
                    target_node=least_loaded[0]
                )

                # 发送任务
                task['trace_id'] = trace_id
                threading.Thread(
                    target=self._send_to_slave,
                    args=(least_loaded[0], least_loaded[1], task, trace_id),
                    daemon=True
                ).start()
            time.sleep(1)

    def _send_to_slave(self, host: str, port: int, task_data: Dict, trace_id: str) -> None:
        """向从节点发送任务"""
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(5)
                s.connect((host, port))
                s.sendall(pickle.dumps(task_data))
                response = s.recv(1024)

                if response == b"ACK":
                    print(Color.colored(f"任务发送成功: {host}", Color.GREEN))
                    self.call_tracer.end_call(trace_id, status='success')
                else:
                    self.call_tracer.end_call(trace_id, status=f"failed: {response.decode()}")
        except Exception as e:
            print(Color.colored(f"发送任务到{host}失败: {e}，将重试...", Color.RED))
            self.call_tracer.end_call(trace_id, status=f"error: {str(e)}")
            # 失败任务重新入队
            with self.lock:
                self.task_queue.append(task_data)

    def _propagate_risk(self, source_ip: str, risk_data: Dict) -> None:
        """风险传播：通知依赖当前机器的节点"""
        if source_ip in self.dependency_graph:
            for dep_ip in self.dependency_graph[source_ip]:
                print(Color.colored(
                    f"风险传播: {source_ip} -> {dep_ip} (风险等级: {risk_data['risk_level']})",
                    Color.YELLOW
                ))
                # 记录风险传播调用链
                trace_id = self.call_tracer.start_chain(
                    service="risk_propagation",
                    operation="alert"
                )
                self.call_tracer.add_call(
                    trace_id=trace_id,
                    service="risk_propagation",
                    operation="send_alert",
                    target_node=dep_ip
                )

                self.send_task_to_node(dep_ip, {
                    'type': 'risk_warning',
                    'source': source_ip,
                    'risk': risk_data,
                    'trace_id': trace_id
                })

                self.call_tracer.end_call(trace_id)

    def add_slave(self, slave_host: str, slave_port: int = 9999) -> None:
        """添加从节点"""
        with self.lock:
            self.slaves.append((slave_host, slave_port))
        print(Color.colored(f"添加从节点: {slave_host}:{slave_port}", Color.CYAN))

    def send_task(self, task_data: Dict) -> None:
        """将任务添加到队列，由任务处理线程分发"""
        with self.lock:
            self.task_queue.append(task_data)

    def send_task_to_node(self, host: str, task_data: Dict) -> None:
        """向指定节点发送任务"""
        for slave_host, slave_port in self.slaves:
            if slave_host == host:
                threading.Thread(
                    target=self._send_to_slave,
                    args=(slave_host, slave_port, task_data, task_data.get('trace_id')),
                    daemon=True
                ).start()
                break

    def stop(self) -> None:
        """停止服务"""
        self.running = False
        if self.server_thread and self.server_thread.is_alive():
            self.server_thread.join()


# 昇腾NPU监控工具类
class AscendNPUMonitor:
    """昇腾NPU监控工具，用于采集NPU相关指标"""

    def __init__(self):
        self.npu_available = self._check_npu()
        self.npu_count = self._get_npu_count() if self.npu_available else 0
        self.npu_thresholds = {
            'usage': {'warning': 70, 'critical': 90},
            'temperature': {'warning': 70, 'critical': 85},
            'memory_usage': {'warning': 75, 'critical': 90}
        }
        self.npu_log_paths = [
            '/var/log/npu/slog/device/',
            '/var/log/npu/driver/',
            '/var/log/npu/scheduler/'
        ]

    def _check_npu(self) -> bool:
        """检查系统是否存在昇腾NPU"""
        try:
            # 尝试调用npu-smi工具
            subprocess.check_output(
                ['npu-smi', '--version'],
                stderr=subprocess.STDOUT,
                timeout=2
            )
            return True
        except (subprocess.SubprocessError, FileNotFoundError):
            return False

    def _get_npu_count(self) -> int:
        """获取NPU设备数量"""
        try:
            output = subprocess.check_output(
                ['npu-smi', 'info'],
                stderr=subprocess.STDOUT,
                timeout=2
            ).decode()
            count_match = re.search(r'Device Count\s*:\s*(\d+)', output)
            return int(count_match.group(1)) if count_match else 0
        except Exception:
            return 0

    def get_npu_metrics(self) -> Dict[str, Any]:
        """采集NPU指标"""
        if not self.npu_available or self.npu_count == 0:
            return {'available': False}

        metrics = {
            'available': True,
            'count': self.npu_count,
            'devices': []
        }

        try:
            for device_id in range(self.npu_count):
                output = subprocess.check_output(
                    ['npu-smi', 'stats', '-i', str(device_id)],
                    stderr=subprocess.STDOUT,
                    timeout=2
                ).decode()

                # 解析使用率
                usage_match = re.search(r'Utilization\s*:\s*(\d+)%', output)
                usage = int(usage_match.group(1)) if usage_match else 0

                # 解析温度
                temp_match = re.search(r'Temperature\s*:\s*(\d+)C', output)
                temperature = int(temp_match.group(1)) if temp_match else 0

                # 解析内存使用
                mem_total_match = re.search(r'Memory Total\s*:\s*(\d+)\s*MiB', output)
                mem_used_match = re.search(r'Memory Used\s*:\s*(\d+)\s*MiB', output)

                mem_total = int(mem_total_match.group(1)) if mem_total_match else 0
                mem_used = int(mem_used_match.group(1)) if mem_used_match else 0
                mem_usage = (mem_used / mem_total) * 100 if mem_total > 0 else 0

                # 解析功耗
                power_match = re.search(r'Power\s*:\s*(\d+\.\d+)\s*W', output)
                power = float(power_match.group(1)) if power_match else 0.0

                # 状态判断
                status = 'normal'
                if (usage >= self.npu_thresholds['usage']['critical'] or
                        temperature >= self.npu_thresholds['temperature']['critical'] or
                        mem_usage >= self.npu_thresholds['memory_usage']['critical']):
                    status = 'critical'
                elif (usage >= self.npu_thresholds['usage']['warning'] or
                      temperature >= self.npu_thresholds['temperature']['warning'] or
                      mem_usage >= self.npu_thresholds['memory_usage']['warning']):
                    status = 'warning'

                metrics['devices'].append({
                    'device_id': device_id,
                    'usage': usage,
                    'temperature': temperature,
                    'memory_used': mem_used,
                    'memory_total': mem_total,
                    'memory_usage': mem_usage,
                    'power': power,
                    'status': status
                })

        except Exception as e:
            print(Color.colored(f"NPU指标采集失败: {e}", Color.RED))

        return metrics

    def check_npu_logs(self) -> List[Dict[str, Any]]:
        """检查NPU相关日志中的错误信息"""
        errors = []
        if not self.npu_available:
            return errors

        for log_path in self.npu_log_paths:
            if not os.path.exists(log_path):
                continue

            try:
                # 只检查最近的日志文件
                log_files = sorted(
                    [f for f in os.listdir(log_path) if os.path.isfile(os.path.join(log_path, f))],
                    key=lambda x: os.path.getmtime(os.path.join(log_path, x)),
                    reverse=True
                )[:5]  # 取最近5个文件

                for file in log_files:
                    file_path = os.path.join(log_path, file)
                    with open(file_path, 'r', errors='ignore') as f:
                        # 查找错误级别日志
                        for line_num, line in enumerate(f, 1):
                            if 'ERROR' in line or 'error' in line or 'CRITICAL' in line:
                                errors.append({
                                    'path': file_path,
                                    'line': line_num,
                                    'content': line.strip(),
                                    'timestamp': datetime.fromtimestamp(os.path.getmtime(file_path))
                                })
            except Exception as e:
                print(Color.colored(f"读取NPU日志失败 {log_path}: {e}", Color.RED))

        return errors

    def analyze_npu_logs(self, lookback_minutes: int = 30) -> List[Dict]:
        """分析NPU日志，检测错误和警告"""
        if not self.npu_available:
            return []

        errors = []
        lookback_time = time.time() - lookback_minutes * 60

        for log_path in self.npu_log_paths:
            if not os.path.exists(log_path):
                continue

            try:
                # 遍历日志目录下的文件
                for filename in os.listdir(log_path):
                    if filename.endswith('.log'):
                        file_path = os.path.join(log_path, filename)
                        # 只处理最近修改的文件
                        if os.path.getmtime(file_path) < lookback_time:
                            continue

                        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                            for line_num, line in enumerate(f):
                                # 检测错误级别日志
                                if 'ERROR' in line or 'error' in line:
                                    errors.append({
                                        'type': 'error',
                                        'path': file_path,
                                        'line': line_num,
                                        'content': line.strip(),
                                        'timestamp': os.path.getmtime(file_path)
                                    })
                                # 检测警告级别日志
                                elif 'WARN' in line or 'warn' in line:
                                    errors.append({
                                        'type': 'warning',
                                        'path': file_path,
                                        'line': line_num,
                                        'content': line.strip(),
                                        'timestamp': os.path.getmtime(file_path)
                                    })
            except Exception as e:
                print(f"{Color.RED}分析NPU日志 {log_path} 失败: {e}{Color.RESET}")

        # 按时间排序
        return sorted(errors, key=lambda x: x['timestamp'], reverse=True)

    def check_npu_anomalies(self) -> List[Dict]:
        """检查NPU异常状态"""
        if not self.npu_available:
            return []

        metrics = self.get_npu_metrics()
        anomalies = []

        for device in metrics.get('devices', []):
            # 检查使用率异常
            if device['usage'] > self.npu_thresholds['usage']['critical']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'usage',
                    'value': device['usage'],
                    'threshold': self.npu_thresholds['usage']['critical'],
                    'level': 'critical',
                    'message': f"NPU {device['device_id']} 使用率过高: {device['usage']}%"
                })
            elif device['usage'] > self.npu_thresholds['usage']['warning']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'usage',
                    'value': device['usage'],
                    'threshold': self.npu_thresholds['usage']['warning'],
                    'level': 'warning',
                    'message': f"NPU {device['device_id']} 使用率较高: {device['usage']}%"
                })

            # 检查温度异常
            if device['temperature'] > self.npu_thresholds['temperature']['critical']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'temperature',
                    'value': device['temperature'],
                    'threshold': self.npu_thresholds['temperature']['critical'],
                    'level': 'critical',
                    'message': f"NPU {device['device_id']} 温度过高: {device['temperature']}°C"
                })
            elif device['temperature'] > self.npu_thresholds['temperature']['warning']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'temperature',
                    'value': device['temperature'],
                    'threshold': self.npu_thresholds['temperature']['warning'],
                    'level': 'warning',
                    'message': f"NPU {device['device_id']} 温度较高: {device['temperature']}°C"
                })

            # 检查内存使用异常
            if device['memory_usage'] > self.npu_thresholds['memory_usage']['critical']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'memory_usage',
                    'value': device['memory_usage'],
                    'threshold': self.npu_thresholds['memory_usage']['critical'],
                    'level': 'critical',
                    'message': f"NPU {device['device_id']} 内存使用率过高: {device['memory_usage']:.1f}%"
                })
            elif device['memory_usage'] > self.npu_thresholds['memory_usage']['warning']:
                anomalies.append({
                    'device_id': device['device_id'],
                    'metric': 'memory_usage',
                    'value': device['memory_usage'],
                    'threshold': self.npu_thresholds['memory_usage']['warning'],
                    'level': 'warning',
                    'message': f"NPU {device['device_id']} 内存使用率较高: {device['memory_usage']:.1f}%"
                })

        # 结合日志异常
        log_errors = self.analyze_npu_logs(lookback_minutes=5)
        for error in log_errors:
            level = 'critical' if error['type'] == 'error' else 'warning'
            anomalies.append({
                'device_id': 'unknown',
                'metric': 'log',
                'value': error['type'],
                'threshold': 'none',
                'level': level,
                'message': f"NPU日志{error['type']}: {error['content']} (文件: {os.path.basename(error['path'])})"
            })

        return anomalies


# 系统监控主类
class SystemMonitor:
    """系统综合监控类，整合CPU、内存、磁盘、网络和NPU监控"""

    def __init__(self, host: str = 'localhost', port: int = 9999, is_master: bool = False,
                 use_trained_models: bool = True, use_database: bool = True):
        self.host = host
        self.port = port
        self.is_master = is_master
        self.use_trained_models = use_trained_models
        self.use_database = use_database

         # 初始化数据库连接
        if self.use_database and MYSQL_AVAILABLE:
            self.db = KylinDatabase()  # 移除硬编码的 password='Huawei@678' 参数
            if not self.db.connect():
                print("警告: 数据库连接失败，将使用本地文件存储")
                self.use_database = False
        
        # 初始化自动故障分析器
        self.auto_fault_analyzer = None
        try:
            from .auto_fault_analyzer import AutoFaultAnalyzer
            self.auto_fault_analyzer = AutoFaultAnalyzer()
            print(f"{Color.GREEN}✅ 自动故障分析器初始化成功{Color.RESET}")
        except Exception as e:
            print(f"{Color.YELLOW}⚠️ 自动故障分析器初始化失败: {e}{Color.RESET}")
            self.auto_fault_analyzer = None
        else:
            if self.use_database and not MYSQL_AVAILABLE:
                #print("警告: MySQL数据库模块不可用，将使用本地文件存储")
                self.use_database = False
            self.db = None

        # 模型训练器 - 使用优化版本以保持特征一致性
        try:
            from detector_model import ModelTrainerOptimized
            self.model_trainer = ModelTrainerOptimized()
        except ImportError:
            # 如果导入失败，使用本地版本
            self.model_trainer = ModelTrainer()
        self.trained_models = {}

        # 异常检测器（在线训练模式）
        self.anomaly_detectors = {
            'isolation_forest': IsolationForest(n_estimators=20),
            'one_class_svm': OneClassSVM(nu=0.1),
            'autoencoder': None  # 动态初始化，需知道输入维度
        }
        self.lstm_predictors = {}  # 按指标类型存储LSTM预测器
        self.feature_history = defaultdict(deque)  # 存储特征历史
        self.history_window = 100  # 历史数据窗口大小
        self.npu_monitor = AscendNPUMonitor()
        self.call_tracer = DistributedCallTracer()

        # 初始化分布式调度器
        if is_master:
            self.scheduler = DistributedScheduler(master_host=host, master_port=port)
        else:
            self.scheduler = None
            self._start_slave_server()

        # 基础阈值配置
        self.base_thresholds = {
            'cpu_usage': {'warning': 70, 'critical': 90},
            'memory_usage': {'warning': 75, 'critical': 90},
            'disk_usage': {'warning': 80, 'critical': 95},
            'network_in': {'warning': 100, 'critical': 200},  # MB/s
            'network_out': {'warning': 100, 'critical': 200},  # MB/s

            # 数据库指标阈值
            'mysql_connections': {'warning': 100, 'critical': 200},
            'mysql_slow_queries': {'warning': 5, 'critical': 10},
            'mysql_threads_connected': {'warning': 80, 'critical': 150},
            'mysql_threads_running': {'warning': 10, 'critical': 20},

            # Redis指标阈值
            'redis_connected_clients': {'warning': 50, 'critical': 100},
            'redis_used_memory': {'warning': 1000000000, 'critical': 2000000000},  # 1GB, 2GB
            'redis_total_commands': {'warning': 10000, 'critical': 50000},

            # Web服务指标阈值
            'nginx_active_connections': {'warning': 100, 'critical': 200},
            'nginx_5xx_errors': {'warning': 5, 'critical': 10},
            'nginx_requests_per_second': {'warning': 100, 'critical': 500},

            'apache_5xx_errors': {'warning': 5, 'critical': 10},
            'apache_requests_per_second': {'warning': 50, 'critical': 200}
        }

        # 场景特定阈值配置
        self.scene_thresholds = {
            'government_peak': {
                'priority': 1.5,
                'cpu_usage': {'warning': 60, 'critical': 80},  # 政务高峰期更敏感
                'memory_usage': {'warning': 65, 'critical': 85},
                'mysql_connections': {'warning': 80, 'critical': 150},
                'nginx_active_connections': {'warning': 80, 'critical': 150}
            },
            'finance_settlement': {
                'priority': 2.0,
                'cpu_usage': {'warning': 65, 'critical': 85},
                'memory_usage': {'warning': 70, 'critical': 90},
                'mysql_connections': {'warning': 60, 'critical': 120},  # 金融结算时更严格
                'mysql_slow_queries': {'warning': 3, 'critical': 8},
                'redis_connected_clients': {'warning': 40, 'critical': 80}
            },
            'energy_monitoring': {
                'priority': 1.8,
                'cpu_usage': {'warning': 75, 'critical': 95},  # 能源监控允许更高负载
                'memory_usage': {'warning': 80, 'critical': 95},
                'network_in': {'warning': 150, 'critical': 300},
                'network_out': {'warning': 150, 'critical': 300}
            },
            'normal': {
                'priority': 1.0
                # 使用基础阈值
            }
        }

        # 当前使用的阈值（初始化为基础阈值）
        self.thresholds = self.base_thresholds.copy()

        # 报告生成相关
        self.report_dir = "data/detection_data"
        self._ensure_report_directory()  # 总是确保目录存在
        self.report_counter = 0
        self.anomaly_history = deque(maxlen=1000)  # 异常历史记录

        # 尝试加载训练好的模型
        if self.use_trained_models:
            self._load_trained_models()

    def _load_trained_models(self):
        """加载训练好的模型"""
        try:
            # 优先从数据库加载模型
            if self.use_database and self.db and MYSQL_AVAILABLE:
                models_loaded = 0
                for model_name in ['isolation_forest', 'one_class_svm', 'autoencoder', 'ensemble_detector']:
                    model = self.db.load_model(model_name)
                    if model:
                        self.trained_models[model_name] = model
                        models_loaded += 1
                        print(f"{Color.GREEN}✓ 从数据库加载模型: {model_name}{Color.RESET}")

                if models_loaded > 0:
                    print(f"{Color.GREEN}成功从数据库加载 {models_loaded} 个模型{Color.RESET}")
                    return

            # 从本地文件加载模型（备用）
            self.trained_models = self.model_trainer.load_models()
            print(f"{Color.GREEN}成功从本地文件加载训练好的模型{Color.RESET}")

        except FileNotFoundError:
            print(f"{Color.YELLOW}没有找到训练好的模型，将使用在线训练模式{Color.RESET}")
            self.use_trained_models = False
        except Exception as e:
            print(f"{Color.RED}加载模型失败: {e}，将使用在线训练模式{Color.RESET}")
            self.use_trained_models = False

    def collect_training_data(self, metrics: Dict):
        """收集训练数据"""
        if not self.use_trained_models:
            self.model_trainer.collect_training_data(metrics)

    def train_models_if_ready(self):
        """如果数据足够，训练模型"""
        if not self.use_trained_models and self.model_trainer.has_sufficient_data():
            try:
                print(f"{Color.CYAN}数据收集完成，开始训练模型...{Color.RESET}")
                self.model_trainer.train_models()
                self.use_trained_models = True
                self._load_trained_models()
                print(f"{Color.GREEN}模型训练完成，已切换到离线模式{Color.RESET}")
            except Exception as e:
                print(f"{Color.RED}模型训练失败: {e}{Color.RESET}")

    def get_model_status(self) -> Dict:
        """获取模型状态信息"""
        status = {
            'use_trained_models': self.use_trained_models,
            'training_data_count': len(self.model_trainer.training_data),
            'has_sufficient_data': self.model_trainer.has_sufficient_data(),
            'model_versions': self.model_trainer.get_model_info()
        }
        return status

    def _ensure_report_directory(self):
        """确保报告目录存在"""
        if not os.path.exists(self.report_dir):
            os.makedirs(self.report_dir)
            print(f"{Color.GREEN}创建报告目录: {self.report_dir}{Color.RESET}")

    def _get_current_scene(self) -> str:
        """基于时间和负载自动识别场景"""
        hour = datetime.now().hour
        weekday = datetime.now().weekday()

        # 政务高峰期：工作日9:00-11:30, 14:00-16:30
        if weekday < 5 and ((9 <= hour < 12) or (14 <= hour < 17)):
            # 检查是否有历史CPU数据
            if hasattr(self, 'feature_history') and 'cpu_usage' in self.feature_history:
                cpu_history = list(self.feature_history['cpu_usage'])
                if cpu_history:
                    cpu_load = cpu_history[-1] if isinstance(cpu_history[-1], (int, float)) else 0
                    if cpu_load > 60:
                        return 'government_peak'

        # 金融结算时段：工作日16:30-18:00
        if weekday < 5 and (16 <= hour < 18):
            # 检查是否有历史MySQL连接数据
            if hasattr(self, 'feature_history') and 'mysql_connections' in self.feature_history:
                mysql_history = list(self.feature_history['mysql_connections'])
                if mysql_history:
                    mysql_connections = mysql_history[-1] if isinstance(mysql_history[-1], (int, float)) else 0
                    if mysql_connections > 80:
                        return 'finance_settlement'

        # 能源监控系统：全天候高负载
        if hasattr(self, 'feature_history') and 'network_in' in self.feature_history:
            net_history = list(self.feature_history['network_in'])
            if net_history:
                net_load = net_history[-1] if isinstance(net_history[-1], (int, float)) else 0
                if net_load > 40:  # 假设40MB/s以上为高负载
                    return 'energy_monitoring'

        # 默认场景
        return 'normal'

    def _adjust_thresholds_by_scene(self) -> Dict:
        """根据当前场景动态调整阈值"""
        current_scene = self._get_current_scene()
        adjusted_thresholds = self.base_thresholds.copy()

        if current_scene in self.scene_thresholds:
            scene_spec = self.scene_thresholds[current_scene]
            for metric, thresholds in scene_spec.items():
                if metric != 'priority' and metric in adjusted_thresholds:
                    adjusted_thresholds[metric] = thresholds

        return adjusted_thresholds

    def _update_thresholds_for_current_scene(self) -> None:
        """更新当前场景的阈值"""
        self.thresholds = self._adjust_thresholds_by_scene()
        current_scene = self._get_current_scene()
        scene_priority = self.scene_thresholds.get(current_scene, {}).get('priority', 1.0)

        # 打印场景信息（仅在场景变化时）
        if not hasattr(self, '_last_scene') or self._last_scene != current_scene:
            print(f"{Color.CYAN}=== 场景识别 ==={Color.RESET}")
            print(f"当前场景: {current_scene}")
            print(f"场景优先级: {scene_priority}")
            print(f"阈值已调整为场景特定值")
            self._last_scene = current_scene

    def generate_report(self, metrics: Dict, anomalies: List[Dict], risk_level: int) -> str:
        """生成详细的监控报告"""
        timestamp = datetime.now()
        report_id = int(time.time())
        
        report_id = timestamp.strftime('%Y-%m-%d-%H-%M-%S')
        
        report_filename = f"异常检测报告_{report_id}.json"
        report_path = os.path.join(self.report_dir, report_filename)

        # 收集主机信息
        host_info = self._collect_host_info()

        # 收集系统日志片段
        system_logs = self._get_system_logs(lookback_minutes=10)

        # 收集调用链快照
        call_chain_snapshot = self._collect_call_chain_snapshot()

        # 收集NPU日志（如果可用）
        npu_logs = []
        if metrics['npu']['available']:
            npu_logs = self.npu_monitor.analyze_npu_logs(lookback_minutes=10)

        # 构建报告数据
        report_data = {
            "report_id": report_id,
            "timestamp": timestamp.isoformat(),
            "host_info": host_info,
            "metrics": metrics,
            "anomalies": anomalies,
            "risk_level": risk_level,
            "system_logs": system_logs,
            "call_chain_snapshot": call_chain_snapshot,
            "npu_logs": npu_logs,
            "anomaly_details": self._analyze_anomalies(anomalies, metrics),
            "recommendations": self._generate_recommendations(anomalies, risk_level)
        }

        # 保存报告
        try:
            with open(report_path, 'w', encoding='utf-8') as f:
                json.dump(report_data, f, ensure_ascii=False, indent=2, default=str)

            # 同时生成文本格式报告
            text_report_path = os.path.join(self.report_dir, f"异常检测报告_{report_id}.txt")
            self._generate_text_report(report_data, text_report_path)

            # 记录异常历史
            self._record_anomaly_history(anomalies, risk_level, timestamp)

            # 新增：自动故障分析和根因分析
            self._auto_fault_analysis(anomalies, metrics, risk_level, report_id)

            print(f"{Color.GREEN}报告已生成: {report_filename}{Color.RESET}")
            return report_path

        except Exception as e:
            print(f"{Color.RED}生成报告失败: {e}{Color.RESET}")
            return ""

    def _auto_fault_analysis(self, anomalies: List[Dict], metrics: Dict, 
                           risk_level: int, report_id: str) -> None:
        """
        自动故障分析和根因分析
        
        Args:
            anomalies: 检测到的异常列表
            metrics: 系统指标数据
            risk_level: 风险等级
            report_id: 报告ID
        """
        if not self.auto_fault_analyzer:
            print(f"{Color.YELLOW}⚠️ 自动故障分析器未初始化，跳过根因分析{Color.RESET}")
            return
        
        if not anomalies:
            print(f"{Color.CYAN}ℹ️ 未检测到异常，跳过自动故障分析{Color.RESET}")
            return
        
        print(f"{Color.CYAN}🔍 开始自动故障分析...{Color.RESET}")
        
        try:
            # 调用自动故障分析器
            analysis_report_path = self.auto_fault_analyzer.analyze_anomalies(
                anomalies, metrics, risk_level, report_id
            )
            
            if analysis_report_path:
                print(f"{Color.GREEN}✅ 自动故障分析完成，根因分析报告已保存: {analysis_report_path}{Color.RESET}")
            else:
                print(f"{Color.YELLOW}⚠️ 自动故障分析未生成报告{Color.RESET}")
                
        except Exception as e:
            print(f"{Color.RED}❌ 自动故障分析失败: {e}{Color.RESET}")

    def _get_real_ip_address(self) -> str:
        """获取真实的网络IP地址，而不是127.0.0.1"""
        try:
            # 方法1: 通过连接外部地址获取本机IP
            import socket
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                # 连接到一个外部地址（不会实际发送数据）
                s.connect(("8.8.8.8", 80))
                local_ip = s.getsockname()[0]
                if local_ip and local_ip != "127.0.0.1":
                    return local_ip
        except Exception:
            pass
        
        try:
            # 方法2: 通过hostname获取，但排除localhost
            hostname = socket.gethostname()
            ip = socket.gethostbyname(hostname)
            if ip and ip != "127.0.0.1":
                return ip
        except Exception:
            pass
        
        try:
            # 方法3: 遍历网络接口
            import psutil
            for interface, addrs in psutil.net_if_addrs().items():
                for addr in addrs:
                    if addr.family == socket.AF_INET and not addr.address.startswith("127."):
                        return addr.address
        except Exception:
            pass
        
        # 如果所有方法都失败，返回默认值
        return "127.0.0.1"

    def _format_memory_size(self, bytes_size: int) -> str:
        """将字节数格式化为更直观的单位"""
        if bytes_size is None or bytes_size == 0:
            return "0 B"
        
        # 定义单位
        units = ['B', 'KB', 'MB', 'GB', 'TB']
        size = float(bytes_size)
        unit_index = 0
        
        # 循环除以1024直到找到合适的单位
        while size >= 1024 and unit_index < len(units) - 1:
            size /= 1024
            unit_index += 1
        
        # 格式化输出，保留2位小数
        if unit_index == 0:  # 字节
            return f"{int(size)} {units[unit_index]}"
        else:
            return f"{size:.2f} {units[unit_index]}"

    def _collect_host_info(self) -> Dict:
        """收集主机详细信息"""
        try:
            memory_total_bytes = psutil.virtual_memory().total
            return {
                "hostname": socket.gethostname(),
                "ip_address": self._get_real_ip_address(),
                "platform": platform.platform(),
                "system": platform.system(),
                "release": platform.release(),
                "version": platform.version(),
                "machine": platform.machine(),
                "processor": platform.processor(),
                "cpu_count": psutil.cpu_count(),
                "cpu_count_logical": psutil.cpu_count(logical=True),
                "memory_total": memory_total_bytes,
                "memory_total_formatted": self._format_memory_size(memory_total_bytes),
                "boot_time": datetime.fromtimestamp(psutil.boot_time()).isoformat(),
                "uptime": time.time() - psutil.boot_time()
            }
        except Exception as e:
            return {"error": f"收集主机信息失败: {str(e)}"}

    def _collect_call_chain_snapshot(self) -> Dict:
        """收集调用链快照"""
        try:
            return {
                "slow_calls": self.call_tracer.get_slow_calls(threshold=1.0),
                "failed_calls": self.call_tracer.get_failed_calls(),
                "total_chains": len(self.call_tracer.call_chains),
                "current_chain": self.call_tracer.current_chain
            }
        except Exception as e:
            return {"error": f"收集调用链快照失败: {str(e)}"}

    def _analyze_anomalies(self, anomalies: List[Dict], metrics: Dict) -> List[Dict]:
        """详细分析异常信息"""
        detailed_anomalies = []

        for anomaly in anomalies:
            detail = {
                "type": anomaly.get('type', 'unknown'),
                "level": anomaly.get('level', 'unknown'),
                "message": anomaly.get('message', ''),
                "value": anomaly.get('value', 0),
                "threshold": anomaly.get('threshold', 0),
                "timestamp": datetime.now().isoformat(),
                "context": {}
            }

            # 根据异常类型添加上下文信息
            if 'cpu' in anomaly.get('type', ''):
                detail["context"] = {
                    "cpu_usage": metrics['cpu']['usage'],
                    "cpu_cores": metrics['cpu']['cores'],
                    "load_average": metrics['cpu']['load_avg'],
                    "per_core_usage": metrics['cpu']['per_core_usage']
                }
            elif 'memory' in anomaly.get('type', ''):
                detail["context"] = {
                    "memory_usage": metrics['memory']['usage'],
                    "memory_available": metrics['memory']['available'],
                    "swap_usage": metrics['memory']['swap_usage']
                }
            elif 'npu' in anomaly.get('type', ''):
                detail["context"] = {
                    "npu_devices": metrics['npu']['devices'] if metrics['npu']['available'] else []
                }
            elif 'disk' in anomaly.get('type', ''):
                detail["context"] = {
                    "disk_usage": metrics['disk']['usage'],
                    "disk_free": metrics['disk']['free'],
                    "disk_io": {
                        "read_bytes": metrics['disk']['read_bytes'],
                        "write_bytes": metrics['disk']['write_bytes']
                    }
                }

            detailed_anomalies.append(detail)

        return detailed_anomalies

    def _generate_recommendations(self, anomalies: List[Dict], risk_level: int) -> List[str]:
        """生成修复建议"""
        recommendations = []

        if risk_level >= 8:
            recommendations.append("风险等级极高，建议立即检查系统状态并采取紧急措施")

        for anomaly in anomalies:
            if anomaly.get('level') == 'critical':
                if 'cpu' in anomaly.get('type', ''):
                    recommendations.append("CPU使用率过高，建议：1) 检查高负载进程 2) 考虑增加CPU资源 3) 优化应用程序")
                elif 'memory' in anomaly.get('type', ''):
                    recommendations.append("内存使用率过高，建议：1) 检查内存泄漏 2) 增加物理内存 3) 优化内存密集型应用")
                elif 'npu' in anomaly.get('type', ''):
                    recommendations.append("NPU异常，建议：1) 检查NPU驱动状态 2) 重启NPU相关服务 3) 检查硬件连接")
                elif 'disk' in anomaly.get('type', ''):
                    recommendations.append("磁盘空间不足，建议：1) 清理临时文件 2) 归档旧日志 3) 扩展磁盘空间")

        if not recommendations:
            recommendations.append("系统运行正常，建议定期监控关键指标")

        return recommendations

    def _format_table_line(self, content: str, width: int = 78) -> str:
        """格式化表格行，只保留左侧边框"""
        # 计算实际可用宽度（减去左侧边框字符）
        available_width = width - 1  # 只减去左侧边框字符
        if len(content) > available_width:
            content = content[:available_width-3] + "..."
        return f"│ {content:<{available_width}}"

    def _generate_text_report(self, report_data: Dict, filepath: str):
        """生成优化的Unicode表格格式报告"""
        try:
            with open(filepath, 'w', encoding='utf-8') as f:
                # 报告头部
                f.write("╔" + "═" * 78 + "╗\n")
                f.write("║" + " " * 34 + "系统监控报告" + " " * 34 + "\n")
                f.write("╚" + "═" * 78 + "╝\n\n")

                # 基本信息
                f.write("📋 基本信息\n")
                f.write("┌" + "─" * 78 + "\n")
                f.write(self._format_table_line(f"报告ID: {report_data['report_id']}"))
                f.write("\n")
                f.write(self._format_table_line(f"生成时间: {report_data['timestamp']}"))
                f.write("\n")
                risk_level = report_data['risk_level']
                risk_icon = "🔴" if risk_level >= 8 else "🟡" if risk_level >= 5 else "🟢"
                f.write(self._format_table_line(f"风险等级: {risk_icon} {risk_level}/10"))
                f.write("\n")
                f.write("└" + "─" * 78 + "\n\n")

                # 异常信息（调整到第二位置）
                f.write("⚠️  异常检测结果\n")
                f.write("┌" + "─" * 78 + "\n")
                if report_data['anomalies']:
                    for i, anomaly in enumerate(report_data['anomalies'], 1):
                        level = anomaly['level'].upper()
                        level_icon = "🔴" if level == "CRITICAL" else "🟡" if level == "WARNING" else "🔵"
                        message = anomaly['message']
                        f.write(self._format_table_line(f"{i:2d}. {level_icon} [{level:<8}] {message}"))
                        f.write("\n")
                else:
                    f.write(self._format_table_line("✅ 未检测到异常"))
                    f.write("\n")
                f.write("└" + "─" * 78 + "\n\n")

                # 主机信息（调整到第三位置）
                f.write("🖥️  主机信息\n")
                f.write("┌" + "─" * 78 + "\n")
                host_info = report_data['host_info']
                f.write(self._format_table_line(f"主机名: {host_info.get('hostname', 'N/A')}"))
                f.write("\n")
                f.write(self._format_table_line(f"IP地址: {host_info.get('ip_address', 'N/A')}"))
                f.write("\n")
                f.write(self._format_table_line(f"操作系统: {host_info.get('platform', 'N/A')}"))
                f.write("\n")
                f.write(self._format_table_line(f"CPU核心数: {host_info.get('cpu_count', 'N/A')}"))
                f.write("\n")
                memory_display = host_info.get('memory_total_formatted', 'N/A')
                f.write(self._format_table_line(f"内存总量: {memory_display}"))
                f.write("\n")
                f.write(self._format_table_line(f"系统启动时间: {host_info.get('boot_time', 'N/A')}"))
                f.write("\n")
                f.write("└" + "─" * 78 + "\n\n")

                # 系统指标摘要
                f.write("📊 系统指标摘要\n")
                f.write("┌" + "─" * 78 + "\n")
                metrics = report_data['metrics']
                
                # CPU指标
                cpu_usage = metrics['cpu']['usage']
                cpu_icon = "🔴" if cpu_usage >= 90 else "🟡" if cpu_usage >= 70 else "🟢"
                f.write(self._format_table_line(f"CPU使用率: {cpu_icon} {cpu_usage:6.2f}%"))
                f.write("\n")
                
                # 内存指标
                memory_usage = metrics['memory']['usage']
                memory_icon = "🔴" if memory_usage >= 90 else "🟡" if memory_usage >= 70 else "🟢"
                f.write(self._format_table_line(f"内存使用率: {memory_icon} {memory_usage:6.2f}%"))
                f.write("\n")
                
                # 磁盘指标
                disk_usage = metrics['disk']['usage']
                disk_icon = "🔴" if disk_usage >= 90 else "🟡" if disk_usage >= 80 else "🟢"
                f.write(self._format_table_line(f"磁盘使用率: {disk_icon} {disk_usage:6.2f}%"))
                f.write("\n")
                
                # 网络指标
                f.write(self._format_table_line(f"网络接收: {metrics['network']['recv_rate']:8.2f} MB/s"))
                f.write("\n")
                f.write(self._format_table_line(f"网络发送: {metrics['network']['sent_rate']:8.2f} MB/s"))
                f.write("\n")

                # NPU状态
                if metrics['npu']['available']:
                    f.write(self._format_table_line("NPU状态:"))
                    f.write("\n")
                    for i, device in enumerate(metrics['npu']['devices']):
                        npu_usage = device['usage']
                        npu_icon = "🔴" if npu_usage >= 90 else "🟡" if npu_usage >= 70 else "🟢"
                        f.write(self._format_table_line(f"  NPU{i}: {npu_icon} 使用率{npu_usage:3d}%, 温度{device['temperature']:3d}°C"))
                        f.write("\n")
                else:
                    f.write(self._format_table_line("NPU状态: 不可用"))
                    f.write("\n")
                
                f.write("└" + "─" * 78 + "\n\n")

                # 调用链信息（调整到倒数第二位置）
                f.write("🔗 调用链快照\n")
                f.write("┌" + "─" * 78 + "\n")
                call_chain = report_data['call_chain_snapshot']
                f.write(self._format_table_line(f"慢调用数量: {len(call_chain.get('slow_calls', []))}"))
                f.write("\n")
                f.write(self._format_table_line(f"失败调用数量: {len(call_chain.get('failed_calls', []))}"))
                f.write("\n")
                f.write(self._format_table_line(f"总调用链数量: {call_chain.get('total_chains', 0)}"))
                f.write("\n")
                f.write("└" + "─" * 78 + "\n\n")

                # 修复建议（调整到最后位置）
                f.write("💡 修复建议\n")
                f.write("┌" + "─" * 78 + "\n")
                for i, rec in enumerate(report_data['recommendations'], 1):
                    f.write(self._format_table_line(f"{i:2d}. {rec}"))
                    f.write("\n")
                f.write("└" + "─" * 78 + "\n\n")

                # 报告尾部
                f.write("╔" + "═" * 78 + "╗\n")
                f.write("║" + " " * 34 + "报告结束    " + " " * 34 + "\n")
                f.write("╚" + "═" * 78 + "╝\n")

        except Exception as e:
            print(f"{Color.RED}生成文本报告失败: {e}{Color.RESET}")

    def _record_anomaly_history(self, anomalies: List[Dict], risk_level: int, timestamp: datetime):
        """记录异常历史"""
        if anomalies:
            self.anomaly_history.append({
                "timestamp": timestamp.isoformat(),
                "anomalies": anomalies,
                "risk_level": risk_level,
                "count": len(anomalies)
            })

    def get_anomaly_statistics(self) -> Dict:
        """获取异常统计信息"""
        if not self.anomaly_history:
            return {"total_anomalies": 0, "risk_levels": [], "anomaly_types": {}}

        stats = {
            "total_anomalies": sum(record["count"] for record in self.anomaly_history),
            "risk_levels": [record["risk_level"] for record in self.anomaly_history],
            "anomaly_types": defaultdict(int)
        }

        for record in self.anomaly_history:
            for anomaly in record["anomalies"]:
                anomaly_type = anomaly.get("type", "unknown")
                stats["anomaly_types"][anomaly_type] += 1

        return dict(stats)

    def export_anomaly_history(self, filename: str = None) -> str:
        """导出异常历史到文件"""
        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"anomaly_history_{timestamp}.json"

        filepath = os.path.join(self.report_dir, filename)

        try:
            history_data = {
                "export_time": datetime.now().isoformat(),
                "total_records": len(self.anomaly_history),
                "statistics": self.get_anomaly_statistics(),
                "history": list(self.anomaly_history)
            }

            with open(filepath, 'w', encoding='utf-8') as f:
                json.dump(history_data, f, ensure_ascii=False, indent=2, default=str)

            print(f"{Color.GREEN}异常历史已导出: {filename}{Color.RESET}")
            return filepath

        except Exception as e:
            print(f"{Color.RED}导出异常历史失败: {e}{Color.RESET}")
            return ""

    def _start_slave_server(self):
        """启动从节点服务器，接收主节点任务"""

        def slave_server():
            try:
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                    # 设置SO_REUSEADDR选项，允许重用地址
                    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                    
                    # 在Windows上，还需要设置SO_EXCLUSIVEADDRUSE为0来允许地址重用
                    if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'):
                        s.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 0)
                    
                    s.bind((self.host, self.port))
                    s.listen(5)
                    print(f"{Color.CYAN}从节点服务器启动: {self.host}:{self.port}{Color.RESET}")
                    
                    while True:
                        conn, addr = s.accept()
                        with conn:
                            data = conn.recv(4096)
                            if data:
                                try:
                                    task = pickle.loads(data)
                                    result = self._process_task(task)
                                    # 附加调用链数据
                                    if 'trace_id' in task:
                                        result['call_chain'] = self.call_tracer.serialize_chain(task['trace_id'])
                                    conn.sendall(pickle.dumps(result))
                                except Exception as e:
                                    conn.sendall(f"ERROR: {str(e)}".encode())
            except OSError as e:
                if e.errno == 10048:  # 地址已被使用
                    print(f"{Color.RED}错误: 端口 {self.port} 已被占用，请选择其他端口或等待端口释放{Color.RESET}")
                else:
                    print(f"{Color.CYAN}从节点服务器启动: Address allready in use{Color.RESET}")
                    #print(f"{Color.RED}从节点服务器启动失败: {e}{Color.RESET}")
            except Exception as e:
                print(f"{Color.RED}从节点服务器发生未知错误: {e}{Color.RESET}")

        threading.Thread(target=slave_server, daemon=True).start()
        print(f"{Color.CYAN}从节点启动: {self.host}:{self.port}{Color.RESET}")

    def _process_task(self, task: Dict) -> Dict:
        """处理主节点发送的任务"""
        task_type = task.get('type', 'monitor')
        result = {'status': 'success', 'host': self.host, 'timestamp': time.time()}

        if task_type == 'monitor':
            # 执行监控任务
            metrics = self.get_system_metrics()
            anomalies = self.detect_anomalies(metrics)
            result['metrics'] = metrics
            result['anomalies'] = anomalies
            result['risk_level'] = self._calculate_risk_level(anomalies)

        elif task_type == 'risk_warning':
            # 处理风险警告
            result['action'] = 'risk_mitigation'
            result['message'] = f"收到来自 {task['source']} 的风险警告: {task['risk']}"
            # 执行风险缓解措施
            self._mitigate_risk(task['risk'])

        elif task_type == 'get_logs':
            # 获取日志
            result['logs'] = self._get_system_logs(lookback_minutes=task.get('lookback', 30))

        else:
            result['status'] = 'error'
            result['message'] = f"未知任务类型: {task_type}"

        return result

    def get_system_metrics(self) -> Dict:
        """获取系统综合指标"""
        metrics = {
            'timestamp': time.time(),
            'host': self.host,
            'cpu': self._get_cpu_metrics(),
            'memory': self._get_memory_metrics(),
            'disk': self._get_disk_metrics(),
            'network': self._get_network_metrics(),
            'services': self._get_service_metrics(),
            'npu': self.npu_monitor.get_npu_metrics()
        }

        # 保存到历史记录
        for metric_type, data in metrics.items():
            if metric_type in ['timestamp', 'host']:
                continue
            self.feature_history[metric_type].append(data)
            if len(self.feature_history[metric_type]) > self.history_window:
                self.feature_history[metric_type].popleft()

        return metrics

    def _get_cpu_metrics(self) -> Dict:
        """获取CPU指标"""
        # 使用非阻塞方式获取CPU使用率，减少延迟
        cpu_usage = psutil.cpu_percent(interval=None)  # 非阻塞调用
        per_core_usage = psutil.cpu_percent(percpu=True, interval=None)  # 非阻塞调用

        return {
            'usage': cpu_usage,
            'cores': psutil.cpu_count(logical=False),
            'logical_cores': psutil.cpu_count(logical=True),
            'load_avg': psutil.getloadavg(),
            'per_core_usage': per_core_usage
        }

    def _get_memory_metrics(self) -> Dict:
        """获取内存指标"""
        mem = psutil.virtual_memory()
        swap = psutil.swap_memory()
        return {
            'total': mem.total,
            'available': mem.available,
            'used': mem.used,
            'usage': mem.percent,
            'swap_total': swap.total,
            'swap_used': swap.used,
            'swap_usage': swap.percent
        }

    def _get_disk_metrics(self) -> Dict:
        """获取磁盘指标"""
        disk_usage = psutil.disk_usage('/')
        disk_io = psutil.disk_io_counters()
        return {
            'total': disk_usage.total,
            'used': disk_usage.used,
            'free': disk_usage.free,
            'usage': disk_usage.percent,
            'read_count': disk_io.read_count,
            'write_count': disk_io.write_count,
            'read_bytes': disk_io.read_bytes,
            'write_bytes': disk_io.write_bytes
        }

    def _get_network_metrics(self) -> Dict:
        """获取网络指标"""
        net = psutil.net_io_counters()
        # 计算每秒速率（基于上次测量）
        now = time.time()
        metrics = {
            'bytes_sent': net.bytes_sent,
            'bytes_recv': net.bytes_recv,
            'packets_sent': net.packets_sent,
            'packets_recv': net.packets_recv,
            'errors_in': net.errin,
            'errors_out': net.errout,
            'dropped_in': net.dropin,
            'dropped_out': net.dropout
        }

        # 计算速率
        if 'network' in self.feature_history and self.feature_history['network']:
            last = self.feature_history['network'][-1]
            elapsed = now - last.get('timestamp', now - 1)
            if elapsed > 0:
                metrics['sent_rate'] = (metrics['bytes_sent'] - last['bytes_sent']) / elapsed / 1024 / 1024  # MB/s
                metrics['recv_rate'] = (metrics['bytes_recv'] - last['bytes_recv']) / elapsed / 1024 / 1024  # MB/s
            else:
                metrics['sent_rate'] = 0
                metrics['recv_rate'] = 0
        else:
            metrics['sent_rate'] = 0
            metrics['recv_rate'] = 0

        metrics['timestamp'] = now
        return metrics

    def _get_service_metrics(self) -> Dict:
        """获取关键服务指标（增强版）"""
        services = {
            'mysql': {'running': False, 'connections': 0, 'slow_queries': 0, 'status': 'down'},
            'redis': {'running': False, 'used_memory': 0, 'connected_clients': 0, 'status': 'down'},
            'nginx': {'running': False, 'active_connections': 0, '5xx_errors': 0, 'status': 'down'},
            'apache': {'running': False, '5xx_errors': 0, 'status': 'down'}
        }

        # 检查进程是否运行
        for proc in psutil.process_iter(['name']):
            try:
                name = proc.info['name'].lower()
                if 'mysql' in name:
                    services['mysql']['running'] = True
                elif 'redis' in name:
                    services['redis']['running'] = True
                elif 'nginx' in name:
                    services['nginx']['running'] = True
                elif 'apache' in name or 'httpd' in name:
                    services['apache']['running'] = True
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue

        # 获取详细数据库指标
        services.update(self._get_database_detailed_metrics(services))

        # 获取详细Web服务指标
        services.update(self._get_web_detailed_metrics(services))

        return services

    def _get_database_detailed_metrics(self, services: Dict) -> Dict:
        """获取数据库详细指标"""
        detailed_metrics = {}

        # MySQL详细指标
        if services['mysql']['running']:
            mysql_metrics = self._get_mysql_metrics()
            detailed_metrics['mysql'] = {**services['mysql'], **mysql_metrics}
        else:
            detailed_metrics['mysql'] = services['mysql']

        # Redis详细指标
        if services['redis']['running']:
            redis_metrics = self._get_redis_metrics()
            detailed_metrics['redis'] = {**services['redis'], **redis_metrics}
        else:
            detailed_metrics['redis'] = services['redis']

        return detailed_metrics

    def _get_web_detailed_metrics(self, services: Dict) -> Dict:
        """获取Web服务详细指标"""
        detailed_metrics = {}

        # Nginx详细指标
        if services['nginx']['running']:
            nginx_metrics = self._get_nginx_metrics()
            detailed_metrics['nginx'] = {**services['nginx'], **nginx_metrics}
        else:
            detailed_metrics['nginx'] = services['nginx']

        # Apache详细指标
        if services['apache']['running']:
            apache_metrics = self._get_apache_metrics()
            detailed_metrics['apache'] = {**services['apache'], **apache_metrics}
        else:
            detailed_metrics['apache'] = services['apache']

        return detailed_metrics

    def _get_mysql_metrics(self) -> Dict:
        """获取MySQL详细指标"""
        mysql_metrics = {
            'connections': 0,
            'slow_queries': 0,
            'status': 'up',  # 默认设为up，因为进程检测已经通过
            'uptime': 0,
            'threads_connected': 0,
            'threads_running': 0,
            'queries_per_second': 0
        }

        # 首先检查MySQL端口是否在监听
        mysql_port_check = self._check_mysql_port()
        if not mysql_port_check:
            mysql_metrics['status'] = 'down'
            print(f"{Color.RED}MySQL端口3306未在监听，服务可能异常{Color.RESET}")
            return mysql_metrics

        # 尝试多种MySQL连接方式
        mysql_connected = False

        # 方式1: 尝试无密码连接（root用户）
        try:
            output = subprocess.check_output([
                'mysql', '-u', 'root', '-e',
                'show status like "Threads_connected"; show status like "Slow_queries"; show status like "Uptime"; show status like "Threads_running"; show status like "Questions"'
            ], timeout=2, stderr=subprocess.STDOUT).decode()
            mysql_connected = True
        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            pass

        # 方式2: 尝试使用系统配置文件连接
        if not mysql_connected:
            try:
                output = subprocess.check_output([
                    'mysql', '--defaults-file=/etc/mysql/my.cnf', '-e',
                    'show status like "Threads_connected"; show status like "Slow_queries"; show status like "Uptime"; show status like "Threads_running"; show status like "Questions"'
                ], timeout=2, stderr=subprocess.STDOUT).decode()
                mysql_connected = True
            except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
                pass

        # 方式3: 尝试使用用户主目录下的配置文件
        if not mysql_connected:
            user_config = os.path.expanduser('~/.my.cnf')
            if os.path.exists(user_config):
                try:
                    output = subprocess.check_output([
                        'mysql', '--defaults-file=' + user_config, '-e',
                        'show status like "Threads_connected"; show status like "Slow_queries"; show status like "Uptime"; show status like "Threads_running"; show status like "Questions"'
                    ], timeout=2, stderr=subprocess.STDOUT).decode()
                    mysql_connected = True
                except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
                    pass

        # 方式4: 尝试使用环境变量
        if not mysql_connected:
            try:
                env = os.environ.copy()
                env['MYSQL_PWD'] = 'your_password_here'  # 需要替换为实际密码
                output = subprocess.check_output([
                    'mysql', '-u', 'root', '-e',
                    'show status like "Threads_connected"; show status like "Slow_queries"; show status like "Uptime"; show status like "Threads_running"; show status like "Questions"'
                ], timeout=2, stderr=subprocess.STDOUT, env=env).decode()
                mysql_connected = True
            except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
                pass

        if mysql_connected:
            # 解析连接数
            conn_match = re.search(r'Threads_connected\s+(\d+)', output)
            if conn_match:
                mysql_metrics['connections'] = int(conn_match.group(1))
                mysql_metrics['threads_connected'] = int(conn_match.group(1))

            # 解析慢查询数
            slow_match = re.search(r'Slow_queries\s+(\d+)', output)
            if slow_match:
                mysql_metrics['slow_queries'] = int(slow_match.group(1))

            # 解析运行时间
            uptime_match = re.search(r'Uptime\s+(\d+)', output)
            if uptime_match:
                mysql_metrics['uptime'] = int(uptime_match.group(1))

            # 解析运行线程数
            running_match = re.search(r'Threads_running\s+(\d+)', output)
            if running_match:
                mysql_metrics['threads_running'] = int(running_match.group(1))

            # 如果能成功执行mysql命令，说明服务完全正常
            mysql_metrics['status'] = 'up'
            print(f"{Color.GREEN}MySQL连接成功，已获取详细指标{Color.RESET}")
        else:
            # MySQL命令不可用，但进程在运行且端口在监听，说明服务基本正常
            # 只是无法通过命令行工具获取详细指标
            mysql_metrics['status'] = 'up'
            # print(f"{Color.YELLOW}MySQL进程运行中且端口正常，但无法通过mysql命令获取详细指标{Color.RESET}")
            # print(f"{Color.CYAN}建议配置MySQL认证信息或创建配置文件{Color.RESET}")

        return mysql_metrics

    def _get_redis_metrics(self) -> Dict:
        """获取Redis详细指标"""
        redis_metrics = {
            'connected_clients': 0,
            'used_memory': 0,
            'status': 'up',  # 默认设为up，因为进程检测已经通过
            'total_commands_processed': 0,
            'keyspace_hits': 0,
            'keyspace_misses': 0
        }

        try:
            # 尝试使用redis-cli获取信息
            output = subprocess.check_output([
                'redis-cli', 'info'
            ], timeout=2, stderr=subprocess.STDOUT).decode()

            # 解析连接客户端数
            clients_match = re.search(r'connected_clients:(\d+)', output)
            if clients_match:
                redis_metrics['connected_clients'] = int(clients_match.group(1))

            # 解析已用内存
            memory_match = re.search(r'used_memory:(\d+)', output)
            if memory_match:
                redis_metrics['used_memory'] = int(memory_match.group(1))

            # 解析总命令处理数
            commands_match = re.search(r'total_commands_processed:(\d+)', output)
            if commands_match:
                redis_metrics['total_commands_processed'] = int(commands_match.group(1))

            # 解析命中率
            hits_match = re.search(r'keyspace_hits:(\d+)', output)
            if hits_match:
                redis_metrics['keyspace_hits'] = int(hits_match.group(1))

            misses_match = re.search(r'keyspace_misses:(\d+)', output)
            if misses_match:
                redis_metrics['keyspace_misses'] = int(misses_match.group(1))

            # 如果能成功执行redis-cli命令，说明服务完全正常
            redis_metrics['status'] = 'up'

        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            # Redis命令不可用，但进程在运行，说明服务基本正常
            # 只是无法通过命令行工具获取详细指标
            redis_metrics['status'] = 'up'
            print(f"{Color.YELLOW}Redis进程运行中，但无法通过redis-cli命令获取详细指标{Color.RESET}")

        return redis_metrics

    def _check_mysql_port(self) -> bool:
        """检查MySQL端口是否在监听"""
        try:
            # 检查3306端口是否在监听
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(1)
            result = sock.connect_ex(('localhost', 3306))
            sock.close()
            return result == 0
        except Exception:
            return False

    def _get_nginx_metrics(self) -> Dict:
        """获取Nginx详细指标"""
        nginx_metrics = {
            'active_connections': 0,
            '5xx_errors': 0,
            'status': 'down',
            'requests_per_second': 0,
            'reading': 0,
            'writing': 0,
            'waiting': 0
        }

        try:
            # 检查Nginx状态
            status_output = subprocess.check_output([
                'systemctl', 'is-active', 'nginx'
            ], timeout=1).decode().strip()

            if status_output in ['active', 'running']:
                nginx_metrics['status'] = 'up'

                # 尝试获取Nginx状态页面信息
                try:
                    status_page = subprocess.check_output([
                        'curl', '-s', 'http://localhost/nginx_status'
                    ], timeout=2).decode()

                    # 解析活动连接数
                    active_match = re.search(r'Active connections: (\d+)', status_page)
                    if active_match:
                        nginx_metrics['active_connections'] = int(active_match.group(1))

                    # 解析请求统计
                    requests_match = re.search(r'(\d+) (\d+) (\d+)', status_page)
                    if requests_match:
                        nginx_metrics['requests_per_second'] = int(requests_match.group(1))

                except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
                    pass

                # 分析访问日志中的5xx错误
                nginx_metrics['5xx_errors'] = self._count_nginx_5xx_errors()

        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            pass

        return nginx_metrics

    def _get_apache_metrics(self) -> Dict:
        """获取Apache详细指标"""
        apache_metrics = {
            '5xx_errors': 0,
            'status': 'down',
            'requests_per_second': 0,
            'bytes_per_second': 0
        }

        try:
            # 检查Apache状态
            status_output = subprocess.check_output([
                'systemctl', 'is-active', 'apache2'
            ], timeout=1).decode().strip()

            if status_output in ['active', 'running']:
                apache_metrics['status'] = 'up'

                # 分析访问日志中的5xx错误
                apache_metrics['5xx_errors'] = self._count_apache_5xx_errors()

        except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
            # 尝试检查httpd服务
            try:
                status_output = subprocess.check_output([
                    'systemctl', 'is-active', 'httpd'
                ], timeout=1).decode().strip()

                if status_output in ['active', 'running']:
                    apache_metrics['status'] = 'up'
                    apache_metrics['5xx_errors'] = self._count_apache_5xx_errors()
            except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
                pass

        return apache_metrics

    def _count_nginx_5xx_errors(self) -> int:
        """统计Nginx访问日志中的5xx错误"""
        error_count = 0
        log_files = [
            '/var/log/nginx/access.log',
            '/var/log/nginx/access.log.1',
            '/var/log/nginx/error.log'
        ]

        for log_file in log_files:
            if os.path.exists(log_file):
                try:
                    with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                        # 读取最后100行
                        lines = deque(f, maxlen=100)
                        for line in lines:
                            # 检查5xx状态码
                            if re.search(r' 5\d{2} ', line):
                                error_count += 1
                except Exception:
                    continue

        return error_count

    def _count_apache_5xx_errors(self) -> int:
        """统计Apache访问日志中的5xx错误"""
        error_count = 0
        log_files = [
            '/var/log/apache2/access.log',
            '/var/log/apache2/access.log.1',
            '/var/log/httpd/access_log',
            '/var/log/httpd/access_log.1'
        ]

        for log_file in log_files:
            if os.path.exists(log_file):
                try:
                    with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                        # 读取最后100行
                        lines = deque(f, maxlen=100)
                        for line in lines:
                            # 检查5xx状态码
                            if re.search(r' 5\d{2} ', line):
                                error_count += 1
                except Exception:
                    continue

        return error_count

    def _get_system_logs(self, lookback_minutes: int = 30) -> List[Dict]:
        """获取系统关键日志"""
        logs = []
        log_files = [
            '/var/log/syslog',
            '/var/log/messages',
            '/var/log/auth.log'
        ]

        lookback_time = time.time() - lookback_minutes * 60

        for log_file in log_files:
            if not os.path.exists(log_file):
                continue

            try:
                with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                    # 只读取文件的最后1000行
                    lines = deque(f, maxlen=1000)
                    for line in lines:
                        # 简单的时间解析（假设日志格式包含可识别的时间戳）
                        log_time = self._parse_log_timestamp(line)
                        if log_time and log_time >= lookback_time:
                            # 检测错误级别
                            level = 'info'
                            if 'ERROR' in line or 'error' in line.lower():
                                level = 'error'
                            elif 'WARN' in line or 'warning' in line.lower():
                                level = 'warning'
                            elif 'INFO' in line:
                                level = 'info'

                            logs.append({
                                'file': log_file,
                                'level': level,
                                'content': line.strip(),
                                'timestamp': log_time
                            })
            except Exception as e:
                print(f"{Color.RED}读取日志文件 {log_file} 失败: {e}{Color.RESET}")

        return sorted(logs, key=lambda x: x['timestamp'], reverse=True)

    def _parse_log_timestamp(self, log_line: str) -> float:
        """解析日志行中的时间戳（简化版）"""
        # 尝试几种常见的日志时间格式
        formats = [
            '%b %d %H:%M:%S',  # 例如: Jan 01 12:34:56
            '%Y-%m-%d %H:%M:%S',  # 例如: 2023-01-01 12:34:56
            '%Y-%m-%dT%H:%M:%S'  # 例如: 2023-01-01T12:34:56
        ]

        for fmt in formats:
            try:
                # 尝试从行首解析时间
                for i in range(min(20, len(log_line))):
                    try:
                        dt = datetime.strptime(log_line[:i], fmt)
                        # 补充年份（如果格式中没有）
                        if dt.year == 1900:
                            dt = dt.replace(year=datetime.now().year)
                        return dt.timestamp()
                    except ValueError:
                        continue
            except Exception:
                continue

        # 无法解析，返回当前时间
        return time.time()

    def extract_features(self, metrics: Dict) -> Dict:
        """从原始指标中提取特征用于异常检测"""
        features = {}

        # 基础指标特征（固定特征集，确保维度一致）
        features['cpu_usage'] = metrics['cpu']['usage']
        features['cpu_user'] = metrics['cpu'].get('user', 0)
        features['cpu_system'] = metrics['cpu'].get('system', 0)
        features['cpu_idle'] = metrics['cpu'].get('idle', 0)
        features['cpu_iowait'] = metrics['cpu'].get('iowait', 0)
        features['cpu_irq'] = metrics['cpu'].get('irq', 0)
        features['cpu_softirq'] = metrics['cpu'].get('softirq', 0)
        features['cpu_steal'] = metrics['cpu'].get('steal', 0)
        features['cpu_guest'] = metrics['cpu'].get('guest', 0)
        features['memory_usage'] = metrics['memory']['usage']
        features['memory_available'] = metrics['memory'].get('available', 0)
        features['memory_used'] = metrics['memory'].get('used', 0)
        features['memory_free'] = metrics['memory'].get('free', 0)
        features['memory_cached'] = metrics['memory'].get('cached', 0)
        features['memory_buffers'] = metrics['memory'].get('buffers', 0)
        features['memory_swap_used'] = metrics['memory'].get('swap_used', 0)
        features['disk_usage'] = metrics['disk']['usage']
        features['disk_read_bytes'] = metrics['disk'].get('read_bytes', 0)
        features['disk_write_bytes'] = metrics['disk'].get('write_bytes', 0)
        features['disk_read_count'] = metrics['disk'].get('read_count', 0)
        features['disk_write_count'] = metrics['disk'].get('write_count', 0)
        features['disk_read_time'] = metrics['disk'].get('read_time', 0)
        features['disk_write_time'] = metrics['disk'].get('write_time', 0)
        features['network_in'] = metrics['network']['recv_rate']
        features['network_out'] = metrics['network']['sent_rate']
        features['network_packets_in'] = metrics['network'].get('packets_recv', 0)
        features['network_packets_out'] = metrics['network'].get('packets_sent', 0)
        features['network_err_in'] = metrics['network'].get('errin', 0)
        features['network_err_out'] = metrics['network'].get('errout', 0)
        features['network_drop_in'] = metrics['network'].get('dropin', 0)
        features['network_drop_out'] = metrics['network'].get('dropout', 0)

        # 数据库详细指标特征
        features['mysql_connections'] = metrics['services']['mysql']['connections']
        features['mysql_slow_queries'] = metrics['services']['mysql']['slow_queries']
        features['mysql_threads_connected'] = metrics['services']['mysql'].get('threads_connected', 0)
        features['mysql_threads_running'] = metrics['services']['mysql'].get('threads_running', 0)
        features['mysql_uptime'] = metrics['services']['mysql'].get('uptime', 0)
        features['mysql_questions'] = metrics['services']['mysql'].get('questions', 0)
        features['mysql_slow_queries'] = metrics['services']['mysql'].get('slow_queries', 0)
        features['mysql_opens'] = metrics['services']['mysql'].get('opens', 0)
        features['mysql_flush_commands'] = metrics['services']['mysql'].get('flush_commands', 0)

        # Redis详细指标特征
        features['redis_connected_clients'] = metrics['services']['redis'].get('connected_clients', 0)
        features['redis_used_memory'] = metrics['services']['redis'].get('used_memory', 0)
        features['redis_total_commands'] = metrics['services']['redis'].get('total_commands_processed', 0)
        features['redis_keyspace_hits'] = metrics['services']['redis'].get('keyspace_hits', 0)
        features['redis_keyspace_misses'] = metrics['services']['redis'].get('keyspace_misses', 0)
        features['redis_evicted_keys'] = metrics['services']['redis'].get('evicted_keys', 0)
        features['redis_expired_keys'] = metrics['services']['redis'].get('expired_keys', 0)

        # Web服务详细指标特征
        features['nginx_active_connections'] = metrics['services']['nginx'].get('active_connections', 0)
        features['nginx_5xx_errors'] = metrics['services']['nginx'].get('5xx_errors', 0)
        features['nginx_requests_per_second'] = metrics['services']['nginx'].get('requests_per_second', 0)
        features['nginx_4xx_errors'] = metrics['services']['nginx'].get('4xx_errors', 0)
        features['nginx_total_requests'] = metrics['services']['nginx'].get('total_requests', 0)

        features['apache_5xx_errors'] = metrics['services']['apache'].get('5xx_errors', 0)
        features['apache_requests_per_second'] = metrics['services']['apache'].get('requests_per_second', 0)

        # 系统日志相关特征
        features['error_log_count'] = metrics.get('error_log_count', 0)
        features['warning_log_count'] = metrics.get('warning_log_count', 0)
        features['critical_log_count'] = metrics.get('critical_log_count', 0)

        # 服务状态特征（转换为数值）
        features['mysql_running'] = 1 if metrics['services']['mysql']['running'] else 0
        features['redis_running'] = 1 if metrics['services']['redis']['running'] else 0
        features['nginx_running'] = 1 if metrics['services']['nginx']['running'] else 0
        features['apache_running'] = 1 if metrics['services']['apache']['running'] else 0

        # 存储特征历史数据（用于场景识别）
        for key, value in features.items():
            if key not in self.feature_history:
                self.feature_history[key] = deque(maxlen=self.history_window)
            self.feature_history[key].append(value)

        # 添加NPU特征（如果可用）- 限制为最多2个NPU设备
        if metrics['npu']['available']:
            for i, device in enumerate(metrics['npu']['devices'][:2]):  # 最多2个NPU设备
                features[f'npu_{i}_usage'] = device['usage']
                features[f'npu_{i}_temp'] = device['temperature']
                features[f'npu_{i}_mem_usage'] = device['memory_usage']
                features[f'npu_{i}_power'] = device.get('power_usage', 0.0)
                features[f'npu_{i}_frequency'] = device.get('frequency', 0.0)
        else:
            # 如果没有NPU，添加默认值
            for i in range(2):
                features[f'npu_{i}_usage'] = 0.0
                features[f'npu_{i}_temp'] = 0.0
                features[f'npu_{i}_mem_usage'] = 0.0
                features[f'npu_{i}_power'] = 0.0
                features[f'npu_{i}_frequency'] = 0.0

        # 滑动窗口特征（限制数量）
        for key in ['cpu_usage', 'memory_usage', 'disk_usage']:
            history = [h[key] for h in self.feature_history.get('cpu' if key == 'cpu_usage' else
                                                                'memory' if key == 'memory_usage' else 'disk', [])
                       if key in h]
            if len(history) >= 5:  # 至少需要5个历史数据点
                window_feats = FeatureEngineering.sliding_window_features(history)
                for k, v in window_feats.items():
                    features[f"{key}_{k}"] = v
            else:
                # 如果历史数据不足，使用默认值
                features[f"{key}_window_mean"] = features[key]
                features[f"{key}_window_std"] = 0.0
                features[f"{key}_window_max"] = features[key]
                features[f"{key}_window_min"] = features[key]
                features[f"{key}_window_range"] = 0.0
                features[f"{key}_window_trend"] = 0.0

        # 周期性特征（限制数量）
        for key in ['cpu_usage', 'memory_usage']:
            history = [h[key] for h in self.feature_history.get('cpu' if key == 'cpu_usage' else 'memory', [])
                       if key in h]
            if len(history) >= 24:  # 至少需要24个历史数据点
                features[f"{key}_periodic_deviation"] = FeatureEngineering.periodic_feature(
                    features[key], history, period=24)
            else:
                features[f"{key}_periodic_deviation"] = 0.0

        # 交叉特征（限制数量，避免特征爆炸）
        cross_feats = FeatureEngineering.cross_features(features)
        # 只保留最重要的交叉特征
        important_cross_features = [
            'cross_cpu_usage_memory_usage',
            'cross_cpu_usage_disk_usage',
            'cross_memory_usage_disk_usage',
            'cross_mysql_connections_redis_connected_clients',
            'cross_nginx_active_connections_apache_requests_per_second'
        ]
        for key in important_cross_features:
            if key in cross_feats:
                features[key] = cross_feats[key]

        return features

    def detect_anomalies(self, metrics: Dict) -> List[Dict]:
        """检测系统异常"""
        # 更新当前场景的阈值
        self._update_thresholds_for_current_scene()

        features = self.extract_features(metrics)
        anomalies = []

        # 基于阈值的异常检测
        threshold_anomalies = self._detect_threshold_anomalies(features)
        anomalies.extend(threshold_anomalies)

        # 基于机器学习的异常检测
        ml_anomalies = self._detect_ml_anomalies(features)
        anomalies.extend(ml_anomalies)

        # NPU特定异常
        if metrics['npu']['available']:
            npu_anomalies = self.npu_monitor.check_npu_anomalies()
            for anomaly in npu_anomalies:
                anomalies.append({
                    'type': f"npu_{anomaly['metric']}",
                    'message': anomaly['message'],
                    'level': anomaly['level'],
                    'value': anomaly['value'],
                    'threshold': anomaly['threshold']
                })

        # 调用链异常（慢调用检测）
        slow_calls = self.call_tracer.get_slow_calls(threshold=2.0)  # 2秒以上视为慢调用
        for call in slow_calls:
            anomalies.append({
                'type': 'slow_call',
                'message': f"慢调用: {call['service']}.{call['operation']} ({call['nodes']}) 耗时 {call['duration']:.2f}s",
                'level': 'warning' if call['duration'] < 5 else 'critical',
                'value': call['duration'],
                'threshold': 2.0
            })

        # 失败调用检测
        failed_calls = self.call_tracer.get_failed_calls()
        for call in failed_calls:
            anomalies.append({
                'type': 'failed_call',
                'message': f"调用失败: {call['service']}.{call['operation']} ({call['nodes']}) 状态: {call['status']}",
                'level': 'critical',
                'value': call['status'],
                'threshold': 'success'
            })

        # 数据库服务异常检测
        db_anomalies = self._detect_database_anomalies(metrics)
        anomalies.extend(db_anomalies)

        # Web服务异常检测
        web_anomalies = self._detect_web_anomalies(metrics)
        anomalies.extend(web_anomalies)

        return anomalies

    def _detect_database_anomalies(self, metrics: Dict) -> List[Dict]:
        """检测数据库服务异常"""
        anomalies = []
        services = metrics['services']

        # MySQL异常检测
        if services['mysql']['running']:
            mysql = services['mysql']

            # 检查MySQL状态（如果存在status字段）
            if 'status' in mysql and mysql['status'] == 'down':
                anomalies.append({
                    'type': 'mysql_status',
                    'message': 'MySQL服务状态异常 【风险等级: 8】',
                    'level': 'critical',
                    'value': mysql['status'],
                    'threshold': 'up'
                })

            # 检查慢查询
            if mysql['slow_queries'] > 10:
                anomalies.append({
                    'type': 'mysql_slow_queries',
                    'message': f'MySQL慢查询过多: {mysql["slow_queries"]}',
                    'level': 'warning' if mysql['slow_queries'] <= 20 else 'critical',
                    'value': mysql['slow_queries'],
                    'threshold': 10
                })

            # 检查连接数
            if mysql['connections'] > 150:
                anomalies.append({
                    'type': 'mysql_connections',
                    'message': f'MySQL连接数过高: {mysql["connections"]}',
                    'level': 'warning' if mysql['connections'] <= 200 else 'critical',
                    'value': mysql['connections'],
                    'threshold': 150
                })

            # 检查线程运行状态（死锁检测）yy
            if mysql.get('threads_running', 0) > 20:
                anomalies.append({
                    'type': 'mysql_deadlock',
                    'message': f'MySQL线程运行数过高，可能存在死锁: {mysql.get("threads_running", 0)}',
                    'level': 'warning' if mysql.get('threads_running', 0) <= 30 else 'critical',
                    'value': mysql.get('threads_running', 0),
                    'threshold': 20
                })

            # 检查线程连接状态
            if mysql.get('threads_connected', 0) > 80:
                anomalies.append({
                    'type': 'mysql_connection_pool',
                    'message': f'MySQL连接池接近耗尽: {mysql.get("threads_connected", 0)}',
                    'level': 'warning' if mysql.get('threads_connected', 0) <= 120 else 'critical',
                    'value': mysql.get('threads_connected', 0),
                    'threshold': 80
                })
        else:
            anomalies.append({
                'type': 'mysql_service',
                'message': 'MySQL服务未运行',
                'level': 'critical',
                'value': 'not_running',
                'threshold': 'running'
            })

        # Redis异常检测
        if services['redis']['running']:
            redis = services['redis']

            # 检查Redis状态（如果存在status字段）
            if 'status' in redis and redis['status'] == 'down':
                anomalies.append({
                    'type': 'redis_status',
                    'message': 'Redis服务状态异常',
                    'level': 'critical',
                    'value': redis['status'],
                    'threshold': 'up'
                })

            # 检查连接客户端数
            if redis['connected_clients'] > 100:
                anomalies.append({
                    'type': 'redis_clients',
                    'message': f'Redis连接客户端数过多: {redis["connected_clients"]}',
                    'level': 'warning' if redis['connected_clients'] <= 150 else 'critical',
                    'value': redis['connected_clients'],
                    'threshold': 100
                })

            # 检查内存使用
            if redis['used_memory'] > 2000000000:  # 2GB
                anomalies.append({
                    'type': 'redis_memory',
                    'message': f'Redis内存使用过高: {redis["used_memory"] / 1000000000:.2f}GB',
                    'level': 'warning' if redis['used_memory'] <= 3000000000 else 'critical',
                    'value': redis['used_memory'],
                    'threshold': 2000000000
                })
        else:
            anomalies.append({
                'type': 'redis_service',
                'message': 'Redis服务未运行',
                'level': 'critical',
                'value': 'not_running',
                'threshold': 'running'
            })

        return anomalies

    def _detect_web_anomalies(self, metrics: Dict) -> List[Dict]:
        """检测Web服务异常"""
        anomalies = []
        services = metrics['services']

        # Nginx异常检测
        if services['nginx']['running']:
            nginx = services['nginx']

            # 检查Nginx状态
            if nginx['status'] == 'down':
                anomalies.append({
                    'type': 'nginx_status',
                    'message': 'Nginx服务状态异常',
                    'level': 'critical',
                    'value': nginx['status'],
                    'threshold': 'up'
                })

            # 检查5xx错误
            if nginx['5xx_errors'] > 10:
                anomalies.append({
                    'type': 'nginx_5xx_errors',
                    'message': f'Nginx 5xx错误过多: {nginx["5xx_errors"]}',
                    'level': 'warning' if nginx['5xx_errors'] <= 20 else 'critical',
                    'value': nginx['5xx_errors'],
                    'threshold': 10
                })

            # 检查活动连接数
            if nginx['active_connections'] > 200:
                anomalies.append({
                    'type': 'nginx_connections',
                    'message': f'Nginx活动连接数过高: {nginx["active_connections"]}',
                    'level': 'warning' if nginx['active_connections'] <= 300 else 'critical',
                    'value': nginx['active_connections'],
                    'threshold': 200
                })
        else:
            anomalies.append({
                'type': 'nginx_service',
                'message': 'Nginx服务未运行',
                'level': 'critical',
                'value': 'not_running',
                'threshold': 'running'
            })

        # Apache异常检测
        if services['apache']['running']:
            apache = services['apache']

            # 检查Apache状态
            if apache['status'] == 'down':
                anomalies.append({
                    'type': 'apache_status',
                    'message': 'Apache服务状态异常',
                    'level': 'critical',
                    'value': apache['status'],
                    'threshold': 'up'
                })

            # 检查5xx错误
            if apache['5xx_errors'] > 10:
                anomalies.append({
                    'type': 'apache_5xx_errors',
                    'message': f'Apache 5xx错误过多: {apache["5xx_errors"]}',
                    'level': 'warning' if apache['5xx_errors'] <= 20 else 'critical',
                    'value': apache['5xx_errors'],
                    'threshold': 10
                })
        else:
            anomalies.append({
                'type': 'apache_service',
                'message': 'Apache服务未运行',
                'level': 'critical',
                'value': 'not_running',
                'threshold': 'running'
            })

        return anomalies

    def _add_risk_level_to_message(self, message: str, level: str, risk_level: int = None) -> str:
        """为异常消息添加风险等级"""
        if risk_level is None:
            # 根据异常级别设置默认风险等级
            if level == 'critical':
                risk_level = 8
            elif level == 'warning':
                risk_level = 3
            else:
                risk_level = 1

        if '【风险等级:' not in message:
            return f"{message} 【风险等级: {risk_level}】"
        return message

    def _detect_threshold_anomalies(self, features: Dict) -> List[Dict]:
        """基于阈值的异常检测（使用动态调整的阈值）"""
        anomalies = []
        current_scene = self._get_current_scene()
        scene_priority = self.scene_thresholds.get(current_scene, {}).get('priority', 1.0)

        for metric, thresholds in self.thresholds.items():
            if metric not in features:
                continue

            value = features[metric]
            # 将英文指标名转换为中文显示名
            metric_names = {
                'cpu_usage': 'CPU使用率',
                'memory_usage': '内存使用率',
                'disk_usage': '磁盘使用率',
                'network_in': '网络入流量',
                'network_out': '网络出流量'
            }
            display_name = metric_names.get(metric, metric)

            # 计算单个异常的风险等级
            def calculate_anomaly_risk_level(level, value, threshold, scene_priority):
                base_risk = 2 if level == 'warning' else 4
                # 根据超出阈值的程度调整风险等级
                exceed_ratio = (value - threshold) / threshold
                if exceed_ratio > 0.5:
                    risk_bonus = 2
                elif exceed_ratio > 0.2:
                    risk_bonus = 1
                else:
                    risk_bonus = 0
                return min(10, base_risk + risk_bonus + int(scene_priority))

            if value > thresholds['critical']:
                risk_level = calculate_anomaly_risk_level('critical', value, thresholds['critical'], scene_priority)
                message = f"{display_name} 严重超标: {value} (阈值: {thresholds['critical']})"
                anomalies.append({
                    'type': metric,
                    'message': self._add_risk_level_to_message(message, 'critical', risk_level),
                    'level': 'critical',
                    'value': value,
                    'threshold': thresholds['critical'],
                    'scene': current_scene,
                    'scene_priority': scene_priority
                })
            elif value > thresholds['warning']:
                risk_level = calculate_anomaly_risk_level('warning', value, thresholds['warning'], scene_priority)
                message = f"{display_name} 警告: {value} (阈值: {thresholds['warning']})"
                anomalies.append({
                    'type': metric,
                    'message': self._add_risk_level_to_message(message, 'warning', risk_level),
                    'level': 'warning',
                    'value': value,
                    'threshold': thresholds['warning'],
                    'scene': current_scene,
                    'scene_priority': scene_priority
                })

        return anomalies

    def _detect_ml_anomalies(self, features: Dict) -> List[Dict]:
        """基于机器学习的异常检测"""
        # 准备特征向量
        feature_list = sorted(features.keys())
        X = np.array([features[key] for key in feature_list]).reshape(1, -1)

        # 检查特征维度是否与模型期望的维度匹配
        if self.use_trained_models and self.trained_models:
            # 获取第一个模型的期望特征维度（假设所有模型使用相同的特征维度）
            expected_dim = None
            for model_name, model in self.trained_models.items():
                if hasattr(model, 'input_dim'):
                    expected_dim = model.input_dim
                    break
                elif hasattr(model, 'n_features_'):
                    expected_dim = model.n_features_
                    break
                elif hasattr(model, 'weights1'):  # Autoencoder模型
                    expected_dim = model.weights1.shape[0]
                    break

            if expected_dim is not None and X.shape[1] != expected_dim:
                print(
                    f"{Color.YELLOW}特征维度不匹配：当前{X.shape[1]}个特征，模型期望{expected_dim}个特征，跳过离线模型检测{Color.RESET}")
                self.use_trained_models = False

        ml_anomalies = []

        # 使用训练好的模型进行检测
        if self.use_trained_models and self.trained_models:
            try:
                # 使用离线训练的模型
                if 'isolation_forest' in self.trained_models:
                    model = self.trained_models['isolation_forest']
                    if hasattr(model, 'predict') and callable(getattr(model, 'predict', None)):
                        if model.predict(X)[0] == 1:
                            score = model.decision_function(X)[0] if hasattr(model, 'decision_function') else 0.5
                            message = "孤立森林算法检测到异常模式（离线模型）"
                            ml_anomalies.append({
                                'type': 'isolation_forest',
                                'message': self._add_risk_level_to_message(message, 'warning', 3),
                                'level': 'warning',
                                'score': score
                            })

                if 'one_class_svm' in self.trained_models:
                    model = self.trained_models['one_class_svm']
                    if hasattr(model, 'predict') and callable(getattr(model, 'predict', None)):
                        if model.predict(X)[0] == 1:
                            score = model.decision_function(X)[0] if hasattr(model, 'decision_function') else 0.5
                            ml_anomalies.append({
                                'type': 'one_class_svm',
                                'message': "单类SVM检测到异常模式（离线模型） 【风险等级: 3】",
                                'level': 'warning',
                                'score': score
                            })

                if 'autoencoder' in self.trained_models:
                    model = self.trained_models['autoencoder']
                    if hasattr(model, 'predict') and callable(getattr(model, 'predict', None)):
                        # 计算重构误差 - 兼容两种自编码器类型
                        try:
                            # 对输入数据进行标准化（与训练时保持一致）
                            try:
                                from sklearn.preprocessing import StandardScaler
                                # 这里应该使用训练时保存的scaler，但为了简化，我们重新计算
                                X_mean = np.mean(X, axis=0)
                                X_std = np.std(X, axis=0) + 1e-6
                                X_normalized = (X - X_mean) / X_std
                                X = X_normalized
                            except ImportError:
                                pass  # 如果sklearn不可用，使用原始数据

                            # 直接使用predict方法，避免复杂的重构误差计算
                            anomaly_score = model.predict(X)[0]

                            if anomaly_score == 1:  # 检测到异常
                                ml_anomalies.append({
                                    'type': 'autoencoder',
                                    'message': f"自编码器检测到异常模式（离线模型） 【风险等级: 3】",
                                    'level': 'warning',
                                    'score': 1.0
                                })
                        except Exception as e:
                            print(f"自编码器检测失败: {e}")
                            # 尝试简单的重构误差计算
                            try:
                                if hasattr(model, 'reconstruct'):
                                    reconstruction = model.reconstruct(X)
                                    # 使用更稳定的误差计算
                                    reconstruction_error = np.mean(np.abs(X - reconstruction))
                                    if reconstruction_error > 0.5:  # 使用更低的阈值
                                        ml_anomalies.append({
                                            'type': 'autoencoder',
                                            'message': f"自编码器检测到异常模式（离线模型），重构误差: {reconstruction_error:.3f} 【风险等级: 3】",
                                            'level': 'warning',
                                            'score': reconstruction_error
                                        })
                            except Exception as e2:
                                print(f"自编码器重构误差计算也失败: {e2}")


                        except Exception as e:
                            print(f"自编码器重构误差计算失败: {e}")
                            # 使用predict方法作为备选
                            try:
                                anomaly_score = model.predict(X)[0]
                                if anomaly_score == 1:  # 检测到异常
                                    ml_anomalies.append({
                                        'type': 'autoencoder',
                                        'message': f"自编码器检测到异常模式（离线模型） 【风险等级: 3】",
                                        'level': 'warning',
                                        'score': 1.0
                                    })
                            except Exception as e2:
                                print(f"自编码器predict方法也失败: {e2}")

                return ml_anomalies

            except Exception as e:
                print(f"{Color.RED}离线模型检测失败: {e}，切换到在线模式{Color.RESET}")
                self.use_trained_models = False

        # 在线训练模式（原有逻辑）
        # 初始化Autoencoder（如果尚未初始化）
        if self.anomaly_detectors['autoencoder'] is None:
            self.anomaly_detectors['autoencoder'] = Autoencoder(input_dim=X.shape[1], epochs=20)

        # 收集足够历史数据后再进行训练和检测
        if len(self.feature_history['cpu']) > self.history_window // 2:
            # 准备训练数据
            train_data = []
            for i in range(len(self.feature_history['cpu'])):
                try:
                    # 构建历史特征向量
                    hist_metrics = {
                        'cpu': self.feature_history['cpu'][i],
                        'memory': self.feature_history['memory'][i],
                        'disk': self.feature_history['disk'][i],
                        'network': self.feature_history['network'][i],
                        'services': self.feature_history['services'][i],
                        'npu': self.feature_history.get('npu', [{}])[i] if i < len(
                            self.feature_history.get('npu', [])) else {}
                    }
                    hist_features = self.extract_features(hist_metrics)
                    train_data.append([hist_features[key] for key in feature_list])
                except Exception:
                    continue

            if len(train_data) >= 10:  # 至少需要10个样本
                # 训练模型
                train_data = np.array(train_data)
                self.anomaly_detectors['isolation_forest'].fit(train_data)
                self.anomaly_detectors['one_class_svm'].fit(train_data)
                self.anomaly_detectors['autoencoder'].fit(train_data)

                # 检测异常
                if self.anomaly_detectors['isolation_forest'].predict(X)[0] == 1:
                    ml_anomalies.append({
                        'type': 'isolation_forest',
                        'message': "孤立森林算法检测到异常模式（在线训练） 【风险等级: 2】",
                        'level': 'warning',
                        'score': self.anomaly_detectors['isolation_forest'].decision_function(X)[0]
                    })

                if self.anomaly_detectors['one_class_svm'].predict(X)[0] == 1:
                    ml_anomalies.append({
                        'type': 'one_class_svm',
                        'message': "单类SVM算法检测到异常模式 【风险等级: 2】",
                        'level': 'warning',
                        'score': self.anomaly_detectors['one_class_svm']._decision_function(X)[0]
                    })

                if self.anomaly_detectors['autoencoder'].predict(X)[0] == 1:
                    ml_anomalies.append({
                        'type': 'autoencoder',
                        'message': "自编码器检测到异常重构误差",
                        'level': 'warning',
                        'score': np.mean(np.square(X - self.anomaly_detectors['autoencoder'].reconstruct(X)))
                    })

        return ml_anomalies

    def _calculate_risk_level(self, anomalies: List[Dict]) -> int:
        """计算总体风险等级（0-10）"""
        if not anomalies:
            return 0

        risk = 0
        for anomaly in anomalies:
            if anomaly['level'] == 'critical':
                risk += 3
            elif anomaly['level'] == 'warning':
                risk += 1

        # 风险等级上限为10
        return min(10, risk)

    def _mitigate_risk(self, risk_data: Dict) -> None:
        """风险缓解措施"""
        risk_level = risk_data.get('risk_level', 0)
        anomalies = risk_data.get('anomalies', [])

        if risk_level < 7:
            return

        print(f"{Color.YELLOW}执行风险缓解措施（风险等级: {risk_level}）{Color.RESET}")

        # 针对不同异常类型执行缓解措施
        for anomaly in anomalies:
            if anomaly['level'] == 'critical':
                if 'cpu_usage' in anomaly['type']:
                    self._mitigate_high_cpu()
                elif 'memory_usage' in anomaly['type']:
                    self._mitigate_high_memory()
                elif 'npu' in anomaly['type']:
                    self._mitigate_npu_issues(anomaly)
                elif 'slow_call' in anomaly['type'] or 'failed_call' in anomaly['type']:
                    self._mitigate_service_issues(anomaly)

    def _mitigate_high_cpu(self) -> None:
        """缓解CPU使用率过高的问题"""
        try:
            # 查找并终止占用CPU过高的进程（简化版）
            high_cpu_procs = []
            for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):
                try:
                    if proc.info['cpu_percent'] > 50:  # CPU使用率超过50%的进程
                        high_cpu_procs.append((proc.info['cpu_percent'], proc.info['pid'], proc.info['name']))
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue

            # 按CPU使用率排序，处理前3个
            for cpu, pid, name in sorted(high_cpu_procs, reverse=True)[:3]:
                print(f"{Color.YELLOW}尝试终止高CPU进程: {name} (PID: {pid}, CPU: {cpu}%){Color.RESET}")
                try:
                    p = psutil.Process(pid)
                    p.terminate()
                    time.sleep(1)
                    if p.is_running():
                        p.kill()
                except Exception as e:
                    print(f"{Color.RED}终止进程 {pid} 失败: {e}{Color.RESET}")
        except Exception as e:
            print(f"{Color.RED}CPU风险缓解失败: {e}{Color.RESET}")

    def _mitigate_high_memory(self) -> None:
        """缓解内存使用率过高的问题"""
        try:
            # 查找并终止占用内存过高的进程
            high_mem_procs = []
            for proc in psutil.process_iter(['pid', 'name', 'memory_percent']):
                try:
                    if proc.info['memory_percent'] > 10:  # 内存使用率超过10%的进程
                        high_mem_procs.append((proc.info['memory_percent'], proc.info['pid'], proc.info['name']))
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue

            # 按内存使用率排序，处理前3个
            for mem, pid, name in sorted(high_mem_procs, reverse=True)[:3]:
                print(f"{Color.YELLOW}尝试终止高内存进程: {name} (PID: {pid}, MEM: {mem}%){Color.RESET}")
                try:
                    p = psutil.Process(pid)
                    p.terminate()
                    time.sleep(1)
                    if p.is_running():
                        p.kill()
                except Exception as e:
                    print(f"{Color.RED}终止进程 {pid} 失败: {e}{Color.RESET}")
        except Exception as e:
            print(f"{Color.RED}内存风险缓解失败: {e}{Color.RESET}")

    def _mitigate_npu_issues(self, anomaly: Dict) -> None:
        """缓解NPU相关问题"""
        if not self.npu_available:
            return

        device_id = anomaly.get('device_id', 'unknown')
        print(f"{Color.YELLOW}尝试缓解NPU {device_id} 问题: {anomaly['message']}{Color.RESET}")

        # 简单的NPU缓解措施（实际环境中可能需要更复杂的逻辑）
        if 'temperature' in anomaly['metric']:
            # 尝试降低NPU负载
            print(f"{Color.CYAN}降低NPU {device_id} 负载以降温{Color.RESET}")
        elif 'usage' in anomaly['metric'] or 'memory_usage' in anomaly['metric']:
            # 尝试重启相关服务释放资源
            print(f"{Color.CYAN}重启NPU {device_id} 相关服务{Color.RESET}")

    def _mitigate_service_issues(self, anomaly: Dict) -> None:
        """缓解服务调用相关问题"""
        print(f"{Color.YELLOW}尝试缓解服务问题: {anomaly['message']}{Color.RESET}")

        # 解析异常信息中的服务名
        service_match = re.search(r'(\w+)\.', anomaly['message'])
        if service_match:
            service_name = service_match.group(1).lower()
            # 尝试重启相关服务
            try:
                if platform.system() == 'Linux':
                    # 假设使用systemd
                    subprocess.run(['systemctl', 'restart', service_name], check=True, timeout=10)
                    print(f"{Color.GREEN}已重启服务: {service_name}{Color.RESET}")
            except Exception as e:
                print(f"{Color.RED}重启服务 {service_name} 失败: {e}{Color.RESET}")

    def map_anomaly_to_core_fault(self, anomaly_type: str, severity: str = 'warning') -> str:
        """将检测到的异常映射到6种核心故障类型"""

        # 直接映射到6种故障类型
        direct_mapping = {
            # CPU相关异常
            'cpu_usage': 'CPU使用率过高（负载过高）',
            'cpu_per_core_usage': 'CPU使用率过高（负载过高）',
            'cpu_temperature': 'CPU使用率过高（负载过高）',
            'cpu_load': 'CPU使用率过高（负载过高）',

            # 内存相关异常
            'memory_usage': '内存泄漏（对象未释放）',
            'memory_available': '内存泄漏（对象未释放）',
            'memory_swap': '内存泄漏（对象未释放）',
            'memory_leak': '内存泄漏（对象未释放）',
            'redis_memory': '内存泄漏（对象未释放）',

            # 磁盘相关异常
            'disk_usage': '磁盘空间耗尽（No space left）',
            'disk_read_bytes': '磁盘空间耗尽（No space left）',
            'disk_write_bytes': '磁盘空间耗尽（No space left）',
            'disk_io_wait': '磁盘空间耗尽（No space left）',
            'disk_space': '磁盘空间耗尽（No space left）',

            # MySQL相关异常
            'mysql_service': 'MySQL死锁（资源竞争）',
            'mysql_connection': 'MySQL死锁（资源竞争）',
            'mysql_deadlock': 'MySQL死锁（资源竞争）',
            'database_connection': 'MySQL死锁（资源竞争）',

            # Web服务器相关异常
            'apache_service': 'Apache线程池耗尽（无法处理新请求）',
            'nginx_service': 'Apache线程池耗尽（无法处理新请求）',
            'apache_threads': 'Apache线程池耗尽（无法处理新请求）',
            'nginx_5xx_errors': 'Apache线程池耗尽（无法处理新请求）',
            'apache_5xx_errors': 'Apache线程池耗尽（无法处理新请求）',

            # 网络相关异常
            'network_bytes_sent': 'TCP拥塞算法切换失败（带宽利用率低）',
            'network_bytes_recv': 'TCP拥塞算法切换失败（带宽利用率低）',
            'network_packets_sent': 'TCP拥塞算法切换失败（带宽利用率低）',
            'network_packets_recv': 'TCP拥塞算法切换失败（带宽利用率低）',
            'tcp_connection': 'TCP拥塞算法切换失败（带宽利用率低）',
        }

        # 如果直接映射存在，直接返回
        if anomaly_type in direct_mapping:
            return direct_mapping[anomaly_type]

        # 智能映射规则（基于关键词）
        anomaly_lower = anomaly_type.lower()

        # CPU相关异常
        if any(keyword in anomaly_lower for keyword in ['cpu', 'load', 'processor', 'core']):
            return 'CPU使用率过高（负载过高）'

        # 内存相关异常
        elif any(keyword in anomaly_lower for keyword in ['memory', 'mem', 'ram', 'swap', 'redis']):
            return '内存泄漏（对象未释放）'

        # 磁盘相关异常
        elif any(keyword in anomaly_lower for keyword in ['disk', 'space', 'storage', 'io', 'filesystem']):
            return '磁盘空间耗尽（No space left）'

        # MySQL相关异常
        elif any(keyword in anomaly_lower for keyword in ['mysql', 'database', 'db', 'sql', 'deadlock', 'lock']):
            return 'MySQL死锁（资源竞争）'

        # Web服务器相关异常
        elif any(
                keyword in anomaly_lower for keyword in ['apache', 'nginx', 'http', 'web', 'thread', 'pool', 'server']):
            return 'Apache线程池耗尽（无法处理新请求）'

        # 网络相关异常
        elif any(keyword in anomaly_lower for keyword in
                 ['network', 'tcp', 'udp', 'bandwidth', 'congestion', 'connection', 'packet']):
            return 'TCP拥塞算法切换失败（带宽利用率低）'

        # 默认映射（根据严重程度）
        else:
            if severity == 'critical':
                return 'CPU使用率过高（负载过高）'
            elif severity == 'warning':
                return '内存泄漏（对象未释放）'
            else:
                return '磁盘空间耗尽（No space left）'

    def analyze_anomaly_with_main(self, anomaly: Dict) -> None:
        """使用main.py分析异常（6种故障类型）"""
        try:
            # 映射到6种核心故障类型
            fault_description = self.map_anomaly_to_core_fault(
                anomaly.get('type', ''),
                anomaly.get('level', 'warning')
            )

            # 调用main.py进行分析
            try:
                import subprocess
                import sys

                # 获取当前脚本所在目录
                current_dir = os.path.dirname(os.path.abspath(__file__))
                main_script = os.path.join(current_dir, 'main.py')

                if os.path.exists(main_script):
                    # 使用subprocess调用main.py
                    result = subprocess.run([
                        sys.executable, main_script,
                        '--fault-description', fault_description
                    ], capture_output=True, text=True, timeout=30)

                    if result.returncode == 0:
                        print(
                            f"{Color.GREEN}✅ 异常已映射并提交分析：{anomaly.get('type')} → {fault_description}{Color.RESET}")
                        if result.stdout.strip():
                            print(f"{Color.CYAN}分析结果：{result.stdout.strip()}{Color.RESET}")
                    else:
                        print(f"{Color.YELLOW}⚠️ main.py执行完成但返回错误：{result.stderr.strip()}{Color.RESET}")
                else:
                    print(f"{Color.YELLOW}⚠️ main.py文件未找到：{main_script}{Color.RESET}")

            except subprocess.TimeoutExpired:
                print(f"{Color.YELLOW}⚠️ main.py分析超时（30秒）{Color.RESET}")
            except Exception as e:
                print(f"{Color.RED}❌ main.py分析失败：{e}{Color.RESET}")

        except Exception as e:
            print(f"{Color.RED}❌ 异常映射失败：{e}{Color.RESET}")

    def test_anomaly_mapping(self) -> None:
        """测试异常映射功能（6种故障类型）"""
        print(f"{Color.CYAN}=== 测试异常映射功能（6种故障类型） ==={Color.RESET}")

        # 测试用例 - 覆盖6种故障类型
        test_cases = [
            # CPU相关异常
            {'type': 'cpu_usage', 'level': 'critical'},
            {'type': 'cpu_per_core_usage', 'level': 'warning'},
            {'type': 'cpu_temperature', 'level': 'critical'},

            # 内存相关异常
            {'type': 'memory_usage', 'level': 'warning'},
            {'type': 'memory_leak', 'level': 'critical'},
            {'type': 'redis_memory', 'level': 'warning'},

            # 磁盘相关异常
            {'type': 'disk_usage', 'level': 'critical'},
            {'type': 'disk_io_wait', 'level': 'warning'},
            {'type': 'disk_space', 'level': 'critical'},

            # MySQL相关异常
            {'type': 'mysql_service', 'level': 'warning'},
            {'type': 'mysql_deadlock', 'level': 'critical'},
            {'type': 'database_connection', 'level': 'warning'},

            # Web服务器相关异常
            {'type': 'apache_service', 'level': 'critical'},
            {'type': 'nginx_5xx_errors', 'level': 'warning'},
            {'type': 'apache_threads', 'level': 'critical'},

            # 网络相关异常
            {'type': 'network_bytes_sent', 'level': 'warning'},
            {'type': 'tcp_connection', 'level': 'critical'},
            {'type': 'network_packets_recv', 'level': 'warning'},

            # 未知异常（测试默认映射）
            {'type': 'unknown_anomaly', 'level': 'warning'},
            {'type': 'other_error', 'level': 'critical'},
        ]

        for test_case in test_cases:
            fault_description = self.map_anomaly_to_core_fault(
                test_case['type'],
                test_case['level']
            )
            print(f"  {test_case['type']} ({test_case['level']}) → {fault_description}")

        print(f"{Color.GREEN}✓ 6种故障类型映射测试完成{Color.RESET}")

        print(f"{Color.GREEN}✅ 异常映射测试完成{Color.RESET}")

    def start_monitoring(self, interval: int = 5, enable_reports: bool = True) -> None:
        """启动监控循环"""
        print(f"{Color.GREEN}开始系统监控，间隔 {interval} 秒{Color.RESET}")
        print(f"{Color.GREEN}报告目录: {os.path.abspath(self.report_dir)}{Color.RESET}")
        if enable_reports:
            print(f"{Color.GREEN}报告生成功能已启用，报告将保存到 {self.report_dir} 目录{Color.RESET}")
        else:
            print(f"{Color.YELLOW}报告生成功能已禁用{Color.RESET}")

        if self.is_master and self.scheduler:
            self.scheduler.start_master()

        try:
            while True:
                start_time = time.time()  # 记录开始时间

                # 主节点定期发送监控任务
                if self.is_master and self.scheduler:
                    self.scheduler.send_task({'type': 'monitor'})

                # 从节点或独立模式下直接监控
                if not self.is_master or (self.is_master and not self.scheduler.slaves):
                    metrics = self.get_system_metrics()

                    # 收集训练数据（如果未使用训练好的模型）
                    self.collect_training_data(metrics)

                    # 检查是否需要训练模型
                    self.train_models_if_ready()

                    anomalies = self.detect_anomalies(metrics)
                    risk_level = self._calculate_risk_level(anomalies)

                    # 打印监控结果
                    self._print_monitoring_results(metrics, anomalies, risk_level, enable_reports)

                # 精确控制间隔时间
                elapsed_time = time.time() - start_time

                #tjut
                sleep_time = max(0, 600 - elapsed_time)
                #sleep_time = max(0, interval - elapsed_time)
                if sleep_time > 0:
                    time.sleep(sleep_time)
                else:
                    # 如果处理时间超过了间隔，打印警告
                    if elapsed_time > interval:
                        print(
                            f"{Color.YELLOW}警告：本次检测耗时 {elapsed_time:.2f} 秒，超过了设定的 {interval} 秒间隔{Color.RESET}")

        except KeyboardInterrupt:
            print(f"\n{Color.CYAN}监控已停止{Color.RESET}")
            if self.is_master and self.scheduler:
                self.scheduler.stop()
    
    def _print_monitoring_results(self, metrics: Dict, anomalies: List[Dict], risk_level: int,
                                  enable_reports: bool = True) -> None:
        """生成监控报告但不打印信息"""
        # 每次都生成报告，无论是否有异常
        if enable_reports:
            self.generate_report(metrics, anomalies, risk_level)        


# 主函数
def main():
    import argparse

    parser = argparse.ArgumentParser(description='分布式系统监控工具')
    parser.add_argument('--host', default='0.0.0.0', help='绑定的主机地址')
    parser.add_argument('--port', type=int, default=9999, help='绑定的端口号')
    parser.add_argument('--master', action='store_true', help='是否作为主节点运行')
    parser.add_argument('--slaves', nargs='*', help='从节点地址列表（主节点模式下）')
    parser.add_argument('--interval', type=int, default=5, help='监控间隔（秒）')
    parser.add_argument('--enable-reports', action='store_true', help='启用自动报告生成')
    parser.add_argument('--export-history', action='store_true', help='导出异常历史并退出')
    parser.add_argument('--no-trained-models', action='store_true', help='禁用训练好的模型，强制使用在线训练模式')
    parser.add_argument('--train-models', action='store_true', help='强制重新训练模型')
    parser.add_argument('--model-status', action='store_true', help='显示模型状态信息并退出')
    parser.add_argument('--test-mapping', action='store_true', help='测试异常映射功能并退出')

    args = parser.parse_args()

    # 初始化监控器
    use_trained_models = not args.no_trained_models
    monitor = SystemMonitor(host=args.host, port=args.port, is_master=args.master,
                            use_trained_models=use_trained_models)

    # 如果只是导出历史
    if args.export_history:
        history_file = monitor.export_anomaly_history()
        if history_file:
            print(f"{Color.GREEN}异常历史已导出到: {history_file}{Color.RESET}")
        else:
            print(f"{Color.YELLOW}没有异常历史可导出{Color.RESET}")
        return

    # 如果只是显示模型状态
    if args.model_status:
        status = monitor.get_model_status()
        # print(f"{Color.CYAN}=== 模型状态信息 ==={Color.RESET}")
        # print(f"使用训练好的模型: {status['use_trained_models']}")
        # print(f"训练数据数量: {status['training_data_count']}")
        # print(f"数据是否足够: {status['has_sufficient_data']}")
        # print(f"模型版本信息: {status['model_versions']}")
        return

    # 如果只是测试异常映射
    if args.test_mapping:
        monitor.test_anomaly_mapping()
        return

    # 如果强制重新训练模型
    if args.train_models:
        if monitor.model_trainer.has_sufficient_data():
            try:
                print(f"{Color.CYAN}强制重新训练模型...{Color.RESET}")
                monitor.model_trainer.train_models()
                monitor.use_trained_models = True
                monitor._load_trained_models()
                print(f"{Color.GREEN}模型重新训练完成！{Color.RESET}")
            except Exception as e:
                print(f"{Color.RED}模型重新训练失败: {e}{Color.RESET}")
        else:
            print(f"{Color.YELLOW}训练数据不足，无法重新训练模型{Color.RESET}")
        return

    # 主节点模式下添加从节点
    if args.master and args.slaves:
        for slave in args.slaves:
            try:
                host, port = slave.split(':')
                monitor.scheduler.add_slave(host, int(port))
            except Exception:
                print(f"{Color.RED}无效的从节点地址: {slave}{Color.RESET}")

    # 启动监控
    monitor.start_monitoring(interval=args.interval, enable_reports=True)  # 默认启用报告生成


if __name__ == "__main__":
    main()
